From a860f2b40c02d2aec945edf1b38e1fcc432b819f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Fri, 19 Jun 2026 00:56:14 +0800 Subject: [PATCH] refactor: narrow storage compatibility surfaces (#3585) * refactor: narrow storage compatibility surfaces * refactor: narrow observability compatibility surfaces --- crates/notify/src/config_manager.rs | 31 ++- crates/notify/src/event.rs | 46 ++-- crates/notify/src/lib.rs | 3 +- crates/notify/src/storage_compat.rs | 108 ++++++++- crates/obs/src/metrics/scheduler.rs | 4 +- crates/obs/src/metrics/stats_collector.rs | 215 ++++------------- crates/obs/src/storage_compat.rs | 221 +++++++++++++++++- crates/s3select-api/src/object_store.rs | 36 +-- crates/s3select-api/src/storage_compat.rs | 28 ++- docs/architecture/migration-progress.md | 134 ++++++++++- rustfs/src/app/object_usecase.rs | 2 +- rustfs/src/server/event.rs | 2 +- scripts/check_architecture_migration_rules.sh | 11 - 13 files changed, 587 insertions(+), 254 deletions(-) diff --git a/crates/notify/src/config_manager.rs b/crates/notify/src/config_manager.rs index 68d183cdd..42e981c98 100644 --- a/crates/notify/src/config_manager.rs +++ b/crates/notify/src/config_manager.rs @@ -13,7 +13,11 @@ // limitations under the License. use crate::{ - Event, NotificationError, registry::TargetRegistry, rule_engine::NotifyRuleEngine, runtime_facade::NotifyRuntimeFacade, + Event, NotificationError, + registry::TargetRegistry, + rule_engine::NotifyRuleEngine, + runtime_facade::NotifyRuntimeFacade, + storage_compat::{self, NotifyConfigStoreError}, }; use rustfs_config::notify::{ NOTIFY_AMQP_SUB_SYS, NOTIFY_KAFKA_SUB_SYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_MYSQL_SUB_SYS, NOTIFY_NATS_SUB_SYS, @@ -316,17 +320,16 @@ impl NotifyConfigManager { where F: FnMut(&mut Config) -> bool, { - let Some(store) = crate::storage_compat::ecstore::global::resolve_object_store_handle() else { - return Err(NotificationError::StorageNotAvailable( - "Failed to save target configuration: server storage not initialized".to_string(), - )); - }; - - let mut new_config = crate::storage_compat::ecstore::config::com::read_config_without_migrate(store.clone()) + let Some(new_config) = storage_compat::update_server_config(&mut modifier) .await - .map_err(|e| NotificationError::ReadConfig(e.to_string()))?; - - if !modifier(&mut new_config) { + .map_err(|err| match err { + NotifyConfigStoreError::StorageNotAvailable => NotificationError::StorageNotAvailable( + "Failed to save target configuration: server storage not initialized".to_string(), + ), + NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err), + NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err), + })? + else { debug!( event = EVENT_NOTIFY_CONFIG_UPDATE, component = LOG_COMPONENT_NOTIFY, @@ -336,11 +339,7 @@ impl NotifyConfigManager { "notify config update" ); return Ok(()); - } - - crate::storage_compat::ecstore::config::com::save_server_config(store, &new_config) - .await - .map_err(|e| NotificationError::SaveConfig(e.to_string()))?; + }; info!( event = EVENT_NOTIFY_CONFIG_UPDATE, diff --git a/crates/notify/src/event.rs b/crates/notify/src/event.rs index df35e9b28..e3eb80a8e 100644 --- a/crates/notify/src/event.rs +++ b/crates/notify/src/event.rs @@ -19,8 +19,6 @@ use rustfs_s3_types::{EventName, event_schema_version}; use serde::{Deserialize, Serialize}; use url::form_urlencoded; -use crate::storage_compat::NotifyObjectInfo; - /// Represents the identity of the user who triggered the event #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -66,6 +64,22 @@ pub struct Object { pub sequencer: String, } +/// Object metadata required by notification event serialization. +#[derive(Debug, Clone, Default)] +pub struct NotifyObjectInfo { + pub bucket: String, + pub name: String, + pub size: i64, + pub etag: Option, + pub content_type: Option, + pub user_defined: HashMap, + pub version_id: Option, + pub mod_time: Option>, + pub restore_expires: Option>, + pub storage_class: Option, + pub transitioned_tier: Option, +} + /// Metadata about the event #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] @@ -206,7 +220,7 @@ impl Event { pub fn new(args: EventArgs) -> Self { let event_time = Utc::now().naive_local(); let sequencer = match args.object.mod_time { - Some(t) => format!("{:X}", t.unix_timestamp_nanos()), + Some(t) => format!("{:X}", t.timestamp_nanos_opt().unwrap_or(0)), None => format!("{:X}", event_time.and_utc().timestamp_nanos_opt().unwrap_or(0)), }; @@ -217,10 +231,7 @@ impl Event { let key_name = form_urlencoded::byte_serialize(args.object.name.as_bytes()).collect::(); let principal_id = args.req_params.get("principalId").unwrap_or(&String::new()).to_string(); - let version_id = match args.object.version_id { - Some(id) => Some(id.to_string()), - None => Some(args.version_id.clone()), - }; + let version_id = args.object.version_id.clone().or_else(|| Some(args.version_id.clone())); let mut s3_metadata = Metadata { schema_version: "1.0".to_string(), @@ -257,11 +268,12 @@ impl Event { } let glacier_event_data = if args.event_name == EventName::ObjectRestoreCompleted { - args.object.restore_expires.and_then(|expiry| { - let expiry_time = DateTime::::from_timestamp(expiry.unix_timestamp(), expiry.nanosecond())?; - let storage_class = args.object.storage_class.clone().or_else(|| { - (!args.object.transitioned_object.tier.is_empty()).then_some(args.object.transitioned_object.tier.clone()) - })?; + args.object.restore_expires.and_then(|expiry_time| { + let storage_class = args + .object + .storage_class + .clone() + .or_else(|| args.object.transitioned_tier.clone())?; Some(GlacierEventData { restore_event_data: RestoreEventData { lifecycle_restoration_expiry_time: expiry_time.to_rfc3339_opts(SecondsFormat::Millis, true), @@ -367,11 +379,11 @@ pub struct EventArgsBuilder { impl EventArgsBuilder { /// Creates a new builder with the required fields. - pub fn new(event_name: EventName, bucket_name: impl Into, object: NotifyObjectInfo) -> Self { + pub fn new(event_name: EventName, bucket_name: impl Into, object: impl Into) -> Self { Self { event_name, bucket_name: bucket_name.into(), - object, + object: object.into(), ..Default::default() } } @@ -389,8 +401,8 @@ impl EventArgsBuilder { } /// Sets the object information. - pub fn object(mut self, object: NotifyObjectInfo) -> Self { - self.object = object; + pub fn object(mut self, object: impl Into) -> Self { + self.object = object.into(); self } @@ -503,7 +515,7 @@ mod tests { NotifyObjectInfo { bucket: "bucket".to_string(), name: "key".to_string(), - restore_expires: Some(time::OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap()), + restore_expires: DateTime::::from_timestamp(1_700_000_000, 0), storage_class: Some("GLACIER".to_string()), ..Default::default() }, diff --git a/crates/notify/src/lib.rs b/crates/notify/src/lib.rs index 9b83662ce..fb92c6576 100644 --- a/crates/notify/src/lib.rs +++ b/crates/notify/src/lib.rs @@ -41,7 +41,7 @@ mod storage_compat; pub use bucket_config_manager::NotifyBucketConfigManager; pub use config_manager::{NotifyConfigManager, runtime_target_id_for_subsystem}; pub use error::{LifecycleError, NotificationError}; -pub use event::{Event, EventArgs, EventArgsBuilder}; +pub use event::{Event, EventArgs, EventArgsBuilder, NotifyObjectInfo}; pub use event_bridge::{LiveEventHistory, NotifyEventBridge}; pub use global::{ initialize, initialize_live_events, is_notification_system_initialized, notification_metrics_snapshot, notification_system, @@ -55,4 +55,3 @@ pub use runtime_facade::NotifyRuntimeFacade; pub use runtime_view::NotifyRuntimeView; pub use services::NotifyServices; pub use status_view::NotifyStatusView; -pub use storage_compat::NotifyObjectInfo; diff --git a/crates/notify/src/storage_compat.rs b/crates/notify/src/storage_compat.rs index be6b0f4e2..15f258334 100644 --- a/crates/notify/src/storage_compat.rs +++ b/crates/notify/src/storage_compat.rs @@ -12,8 +12,110 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) mod ecstore { - pub(crate) use rustfs_ecstore::{config, global}; +use chrono::{DateTime, Utc}; +use rustfs_config::server_config::Config; +use rustfs_ecstore::{config, global}; + +use crate::event::NotifyObjectInfo; + +type EcstoreObjectInfo = rustfs_ecstore::store_api::ObjectInfo; + +#[derive(Debug)] +pub(crate) enum NotifyConfigStoreError { + StorageNotAvailable, + Read(String), + Save(String), } -pub type NotifyObjectInfo = rustfs_ecstore::store_api::ObjectInfo; +pub(crate) async fn update_server_config(mut modifier: F) -> Result, NotifyConfigStoreError> +where + F: FnMut(&mut Config) -> bool, +{ + let Some(store) = global::resolve_object_store_handle() else { + return Err(NotifyConfigStoreError::StorageNotAvailable); + }; + + let mut new_config = config::com::read_config_without_migrate(store.clone()) + .await + .map_err(|err| NotifyConfigStoreError::Read(err.to_string()))?; + + if !modifier(&mut new_config) { + return Ok(None); + } + + config::com::save_server_config(store, &new_config) + .await + .map_err(|err| NotifyConfigStoreError::Save(err.to_string()))?; + + Ok(Some(new_config)) +} + +impl From for NotifyObjectInfo { + fn from(object: EcstoreObjectInfo) -> Self { + Self { + bucket: object.bucket, + name: object.name, + size: object.size, + etag: object.etag, + content_type: object.content_type, + user_defined: object + .user_defined + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect(), + version_id: object.version_id.map(|version_id| version_id.to_string()), + mod_time: object + .mod_time + .and_then(|value| DateTime::::from_timestamp(value.unix_timestamp(), value.nanosecond())), + restore_expires: object + .restore_expires + .and_then(|value| DateTime::::from_timestamp(value.unix_timestamp(), value.nanosecond())), + storage_class: object.storage_class, + transitioned_tier: (!object.transitioned_object.tier.is_empty()).then_some(object.transitioned_object.tier), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::TransitionedObject; + use std::{collections::HashMap, sync::Arc}; + use time::{Duration, OffsetDateTime}; + + #[test] + fn ecstore_object_info_conversion_preserves_notify_event_fields() { + let mod_time = OffsetDateTime::UNIX_EPOCH + Duration::seconds(42); + let restore_expires = OffsetDateTime::UNIX_EPOCH + Duration::seconds(1_700_000_000); + let mut metadata = HashMap::new(); + metadata.insert("x-amz-meta-key".to_string(), "value".to_string()); + + let converted = NotifyObjectInfo::from(EcstoreObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + size: 123, + etag: Some("etag".to_string()), + content_type: Some("text/plain".to_string()), + user_defined: Arc::new(metadata), + mod_time: Some(mod_time), + restore_expires: Some(restore_expires), + storage_class: Some("GLACIER".to_string()), + transitioned_object: TransitionedObject { + tier: "DEEP_ARCHIVE".to_string(), + ..Default::default() + }, + ..Default::default() + }); + + assert_eq!(converted.bucket, "bucket"); + assert_eq!(converted.name, "object"); + assert_eq!(converted.size, 123); + assert_eq!(converted.etag.as_deref(), Some("etag")); + assert_eq!(converted.content_type.as_deref(), Some("text/plain")); + assert_eq!(converted.user_defined.get("x-amz-meta-key").map(String::as_str), Some("value")); + assert_eq!(converted.mod_time, DateTime::::from_timestamp(42, 0)); + assert_eq!(converted.restore_expires, DateTime::::from_timestamp(1_700_000_000, 0)); + assert_eq!(converted.storage_class.as_deref(), Some("GLACIER")); + assert_eq!(converted.transitioned_tier.as_deref(), Some("DEEP_ARCHIVE")); + } +} diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 664a167ec..e23c59218 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -81,7 +81,7 @@ use crate::metrics::stats_collector::{ 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 crate::storage_compat::ecstore::global::get_global_bucket_monitor; +use crate::storage_compat::obs_bucket_monitor_available; use rustfs_audit::audit_target_metrics; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; use rustfs_utils::get_env_opt_u64; @@ -599,7 +599,7 @@ pub fn init_metrics_runtime(token: CancellationToken) { loop { tokio::select! { _ = interval.tick() => { - let monitor_available = get_global_bucket_monitor().is_some(); + let monitor_available = obs_bucket_monitor_available(); let stats = collect_bucket_replication_bandwidth_stats(); let current_live_keys = repl_bw_live_keys(&stats); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index ff203a82e..5a7760cd5 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -21,18 +21,16 @@ //! and convert them to the Stats structs used by collectors. 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, + BucketReplicationBandwidthStats, BucketReplicationStats, BucketStats, BucketUsageStats, ClusterConfigStats, + ClusterHealthStats, ClusterStats, ClusterUsageStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, + ErasureSetStats, HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, + ReplicationStats, ResourceStats, ScannerStats, +}; +use crate::storage_compat::{ + load_obs_data_usage_from_backend, obs_bucket_quota_limit_bytes, obs_bucket_replication_bandwidth_stats, + obs_bucket_replication_detail_stats, obs_ilm_runtime_snapshot, obs_site_replication_stats, obs_total_usable_capacity_bytes, + obs_total_usable_capacity_free_bytes, resolve_obs_object_store_handle, }; -use crate::storage_compat::ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, GLOBAL_TransitionState}; -use crate::storage_compat::ecstore::bucket::metadata_sys::get_quota_config; -use crate::storage_compat::ecstore::bucket::replication::GLOBAL_REPLICATION_STATS; -use crate::storage_compat::ecstore::data_usage::load_data_usage_from_backend; -use crate::storage_compat::ecstore::global::get_global_bucket_monitor; -use crate::storage_compat::ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free}; -use crate::storage_compat::ecstore::resolve_object_store_handle; use chrono::Utc; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::global_metrics; @@ -41,7 +39,6 @@ use rustfs_io_metrics::internode_metrics::global_internode_metrics; use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system}; use rustfs_storage_api::{BucketOperations, BucketOptions, StorageAdminApi}; use std::collections::HashMap; -use std::time::Duration; use sysinfo::{Networks, System}; use tracing::{instrument, warn}; @@ -151,15 +148,15 @@ pub struct ProcessMetricBundle { /// Collect cluster and cluster-health statistics from a single storage snapshot. pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthStats) { - let Some(store) = resolve_object_store_handle() else { + let Some(store) = resolve_obs_object_store_handle() else { return (ClusterStats::default(), ClusterHealthStats::default()); }; let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; let raw_capacity: u64 = storage_info.disks.iter().map(|d| d.total_space).sum(); let used: u64 = storage_info.disks.iter().map(|d| d.used_space).sum(); - let usable_capacity = get_total_usable_capacity(&storage_info.disks, &storage_info) as u64; - let free = get_total_usable_capacity_free(&storage_info.disks, &storage_info) as u64; + let usable_capacity = obs_total_usable_capacity_bytes(&storage_info); + let free = obs_total_usable_capacity_free_bytes(&storage_info); let stale_capacity_drives = storage_info .disks .iter() @@ -178,7 +175,7 @@ pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthS .count() as u64; // Get bucket and object counts from data usage info. - let (buckets_count, objects_count) = match load_data_usage_from_backend(store.clone()).await { + let (buckets_count, objects_count) = match load_obs_data_usage_from_backend(store.clone()).await { Ok(data_usage) => (data_usage.buckets_count, data_usage.objects_total_count), Err(e) => { warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "cluster_stats", result = "data_usage_load_failed", error = %e, "metrics collector state changed"); @@ -241,12 +238,12 @@ pub async fn collect_cluster_health_stats() -> ClusterHealthStats { /// Collect bucket statistics from the storage layer. pub async fn collect_bucket_stats() -> Vec { - let Some(store) = resolve_object_store_handle() else { + let Some(store) = resolve_obs_object_store_handle() else { return Vec::new(); }; // Load data usage info from backend to get bucket sizes and object counts - let data_usage = match load_data_usage_from_backend(store.clone()).await { + let data_usage = match load_obs_data_usage_from_backend(store.clone()).await { Ok(info) => Some(info), Err(e) => { warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "bucket_stats", result = "data_usage_load_failed", error = %e, "metrics collector state changed"); @@ -284,10 +281,7 @@ pub async fn collect_bucket_stats() -> Vec { .unwrap_or((0, 0)); // Get quota from bucket metadata - let quota_bytes = match get_quota_config(&bucket.name).await { - Ok((quota, _)) => quota.get_quota_limit().unwrap_or(0), - Err(_) => 0, // No quota configured or error - }; + let quota_bytes = obs_bucket_quota_limit_bytes(&bucket.name).await; stats.push(BucketStats { name: bucket.name, @@ -302,26 +296,24 @@ pub async fn collect_bucket_stats() -> Vec { /// Collect bucket replication bandwidth stats from the global monitor. pub fn collect_bucket_replication_bandwidth_stats() -> Vec { - let Some(monitor) = get_global_bucket_monitor() else { + let Some(bandwidth_stats) = obs_bucket_replication_bandwidth_stats() else { return Vec::new(); }; - monitor - .get_report(|_| true) - .bucket_stats + bandwidth_stats .into_iter() - .map(|(opts, details)| { - let target_arn = opts.replication_arn; - let limit_bytes_per_sec = u64::try_from(details.limit_bytes_per_sec).unwrap_or_else(|_| { - warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "bucket_replication_bandwidth", result = "invalid_limit_value", target_arn = ?target_arn, limit_value = details.limit_bytes_per_sec, "metrics collector state changed"); + .map(|stat| { + let target_arn = stat.target_arn; + let limit_bytes_per_sec = u64::try_from(stat.limit_bytes_per_sec).unwrap_or_else(|_| { + warn!(event = EVENT_METRICS_COLLECTOR_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_COLLECTOR, collector = "bucket_replication_bandwidth", result = "invalid_limit_value", target_arn = ?target_arn, limit_value = stat.limit_bytes_per_sec, "metrics collector state changed"); 0 }); BucketReplicationBandwidthStats { - bucket: opts.name, + bucket: stat.bucket, target_arn, limit_bytes_per_sec, - current_bandwidth_bytes_per_sec: details.current_bandwidth_bytes_per_sec, + current_bandwidth_bytes_per_sec: stat.current_bandwidth_bytes_per_sec, } }) .collect() @@ -329,127 +321,12 @@ pub fn collect_bucket_replication_bandwidth_stats() -> Vec Vec { - let Some(stats) = GLOBAL_REPLICATION_STATS.get() 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 += target_stats.fail_stats.size.max(0) as u64; - total_failed_count += target_stats.fail_stats.count.max(0) as u64; - - let last_min = target_stats.fail_stats.recent_since(Duration::from_secs(60)); - last_min_failed_bytes += last_min.size.max(0) as u64; - last_min_failed_count += last_min.count.max(0) as u64; - - let last_hour = target_stats.fail_stats.recent_since(Duration::from_secs(60 * 60)); - last_hour_failed_bytes += last_hour.size.max(0) as u64; - last_hour_failed_count += last_hour.count.max(0) as u64; - - sent_bytes += target_stats.replicated_size.max(0) as u64; - sent_count += target_stats.replicated_count.max(0) as u64; - - targets.push(BucketReplicationTargetStats { - target_arn, - bandwidth_limit_bytes_per_sec: target_stats.bandwidth_limit_bytes_per_sec.max(0) as u64, - 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: proxy.get_total.max(0) as u64, - proxied_get_requests_failures: proxy.get_failed.max(0) as u64, - proxied_head_requests_total: proxy.head_total.max(0) as u64, - proxied_head_requests_failures: proxy.head_failed.max(0) as u64, - proxied_put_requests_total: proxy.put_total.max(0) as u64, - proxied_put_requests_failures: proxy.put_failed.max(0) as u64, - proxied_put_tagging_requests_total: proxy.put_tag_total.max(0) as u64, - proxied_put_tagging_requests_failures: proxy.put_tag_failed.max(0) as u64, - proxied_get_tagging_requests_total: proxy.get_tag_total.max(0) as u64, - proxied_get_tagging_requests_failures: proxy.get_tag_failed.max(0) as u64, - proxied_delete_tagging_requests_total: proxy.delete_tag_total.max(0) as u64, - proxied_delete_tagging_requests_failures: proxy.delete_tag_failed.max(0) as u64, - targets, - }); - } - - buckets + obs_bucket_replication_detail_stats().await } /// Collect site-level replication stats from the global replication runtime. pub async fn collect_replication_stats() -> ReplicationStats { - let Some(stats) = GLOBAL_REPLICATION_STATS.get() else { - return ReplicationStats::default(); - }; - - let site_metrics = stats.get_sr_metrics_for_node().await; - let current_active_workers = u64::try_from(site_metrics.active_workers.curr).unwrap_or(0); - - let bandwidth_stats = collect_bucket_replication_bandwidth_stats(); - let current_data_transfer_rate = bandwidth_stats - .iter() - .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::() - .try_into() - .unwrap_or(0); - - 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: current_active_workers, - current_data_transfer_rate, - last_minute_queued_bytes: site_metrics.queued.last_minute.bytes.max(0) as u64, - last_minute_queued_count: site_metrics.queued.last_minute.count.max(0) as u64, - max_active_workers: u64::try_from(site_metrics.active_workers.max).unwrap_or(0), - max_queued_bytes: site_metrics.queued.max.bytes.max(0) as u64, - max_queued_count: site_metrics.queued.max.count.max(0) as u64, - max_data_transfer_rate, - recent_backlog_count, - } + obs_site_replication_stats().await } /// Collect disk statistics from the storage layer. @@ -522,7 +399,7 @@ pub fn collect_system_memory_stats() -> MemoryStats { /// Collect node disk stats and drive stats from a single storage snapshot. pub async fn collect_disk_and_system_drive_stats() -> (Vec, Vec, DriveCountStats) { - let Some(store) = resolve_object_store_handle() else { + let Some(store) = resolve_obs_object_store_handle() else { return (Vec::new(), Vec::new(), DriveCountStats::default()); }; @@ -710,7 +587,7 @@ pub fn collect_internode_network_stats() -> Option { /// Collect cluster config metrics from backend parity configuration. pub async fn collect_cluster_config_stats() -> Option { - let store = resolve_object_store_handle()?; + let store = resolve_obs_object_store_handle()?; let backend = StorageAdminApi::backend_info(store.as_ref()).await; Some(ClusterConfigStats { @@ -721,7 +598,7 @@ pub async fn collect_cluster_config_stats() -> Option { /// Collect cluster erasure set metrics from storage and backend topology info. pub async fn collect_erasure_set_stats() -> Vec { - let Some(store) = resolve_object_store_handle() else { + let Some(store) = resolve_obs_object_store_handle() else { return Vec::new(); }; @@ -800,8 +677,8 @@ pub async fn collect_iam_stats() -> Option { /// builds cluster totals plus per-bucket distributions from the returned /// histograms. It does not trigger an inline object-data rescan. pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, Vec)> { - let store = resolve_object_store_handle()?; - let data_usage = load_data_usage_from_backend(store.clone()).await.ok()?; + let store = resolve_obs_object_store_handle()?; + let data_usage = load_obs_data_usage_from_backend(store.clone()).await.ok()?; let mut buckets = Vec::with_capacity(data_usage.buckets_usage.len()); for (bucket_name, usage) in &data_usage.buckets_usage { @@ -809,10 +686,7 @@ pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, continue; } - let quota_bytes = match get_quota_config(bucket_name).await { - Ok((quota, _)) => quota.get_quota_limit().unwrap_or(0), - Err(_) => 0, - }; + let quota_bytes = obs_bucket_quota_limit_bytes(bucket_name).await; buckets.push(BucketUsageStats { bucket: bucket_name.clone(), @@ -869,26 +743,19 @@ pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, /// Collect ILM metrics from the current lifecycle runtime state. pub async fn collect_ilm_metric_stats() -> Option { - let expiry_pending_tasks = GLOBAL_ExpiryState.read().await.pending_tasks() as u64; - let transition_active_tasks = GLOBAL_TransitionState.active_tasks().max(0) as u64; - let transition_pending_tasks = GLOBAL_TransitionState.pending_tasks() as u64; - let transition_missed_immediate_tasks = GLOBAL_TransitionState.missed_immediate_tasks().max(0) as u64; - let transition_queue_full_tasks = GLOBAL_TransitionState.queue_full_tasks().max(0) as u64; - let transition_queue_send_timeout_tasks = GLOBAL_TransitionState.queue_send_timeout_tasks().max(0) as u64; - let transition_compensation_scheduled_tasks = GLOBAL_TransitionState.compensation_scheduled_tasks().max(0) as u64; - let transition_compensation_running_tasks = GLOBAL_TransitionState.compensation_running_tasks().max(0) as u64; + let ilm = obs_ilm_runtime_snapshot().await; let metrics = global_metrics().report().await; let versions_scanned = metrics.life_time_ilm.values().copied().sum(); Some(IlmStats { - expiry_pending_tasks, - transition_active_tasks, - transition_pending_tasks, - transition_missed_immediate_tasks, - transition_queue_full_tasks, - transition_queue_send_timeout_tasks, - transition_compensation_scheduled_tasks, - transition_compensation_running_tasks, + expiry_pending_tasks: ilm.expiry_pending_tasks, + transition_active_tasks: ilm.transition_active_tasks, + transition_pending_tasks: ilm.transition_pending_tasks, + transition_missed_immediate_tasks: ilm.transition_missed_immediate_tasks, + transition_queue_full_tasks: ilm.transition_queue_full_tasks, + transition_queue_send_timeout_tasks: ilm.transition_queue_send_timeout_tasks, + transition_compensation_scheduled_tasks: ilm.transition_compensation_scheduled_tasks, + transition_compensation_running_tasks: ilm.transition_compensation_running_tasks, versions_scanned, }) } diff --git a/crates/obs/src/storage_compat.rs b/crates/obs/src/storage_compat.rs index 7e2ff918c..3c2e60461 100644 --- a/crates/obs/src/storage_compat.rs +++ b/crates/obs/src/storage_compat.rs @@ -12,6 +12,223 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) mod ecstore { - pub(crate) use rustfs_ecstore::{bucket, data_usage, global, pools, resolve_object_store_handle}; +use crate::metrics::collectors::{ + BucketReplicationStats as MetricBucketReplicationStats, BucketReplicationTargetStats, + ReplicationStats as MetricReplicationStats, +}; +use rustfs_storage_api::StorageAdminApi; +use std::sync::Arc; +use std::time::Duration; + +pub(crate) use rustfs_ecstore::data_usage::load_data_usage_from_backend as load_obs_data_usage_from_backend; + +pub(crate) type ObsStore = rustfs_ecstore::store::ECStore; +type ObsStorageInfo = ::StorageInfo; + +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct ObsBucketReplicationBandwidthStats { + pub(crate) bucket: String, + pub(crate) target_arn: String, + pub(crate) limit_bytes_per_sec: i64, + pub(crate) current_bandwidth_bytes_per_sec: f64, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct ObsIlmRuntimeSnapshot { + pub(crate) expiry_pending_tasks: u64, + pub(crate) transition_active_tasks: u64, + pub(crate) transition_pending_tasks: u64, + pub(crate) transition_missed_immediate_tasks: u64, + pub(crate) transition_queue_full_tasks: u64, + pub(crate) transition_queue_send_timeout_tasks: u64, + pub(crate) transition_compensation_scheduled_tasks: u64, + pub(crate) transition_compensation_running_tasks: u64, +} + +pub(crate) fn resolve_obs_object_store_handle() -> Option> { + rustfs_ecstore::resolve_object_store_handle() +} + +pub(crate) fn obs_total_usable_capacity_bytes(storage_info: &ObsStorageInfo) -> u64 { + usize_to_u64_saturating(rustfs_ecstore::pools::get_total_usable_capacity(&storage_info.disks, storage_info)) +} + +pub(crate) fn obs_total_usable_capacity_free_bytes(storage_info: &ObsStorageInfo) -> u64 { + usize_to_u64_saturating(rustfs_ecstore::pools::get_total_usable_capacity_free(&storage_info.disks, storage_info)) +} + +pub(crate) async fn obs_bucket_quota_limit_bytes(bucket: &str) -> u64 { + rustfs_ecstore::bucket::metadata_sys::get_quota_config(bucket) + .await + .ok() + .and_then(|(quota, _)| quota.get_quota_limit()) + .unwrap_or(0) +} + +pub(crate) fn obs_bucket_monitor_available() -> bool { + rustfs_ecstore::global::get_global_bucket_monitor().is_some() +} + +pub(crate) fn obs_bucket_replication_bandwidth_stats() -> Option> { + let monitor = rustfs_ecstore::global::get_global_bucket_monitor()?; + Some( + monitor + .get_report(|_| true) + .bucket_stats + .into_iter() + .map(|(opts, details)| ObsBucketReplicationBandwidthStats { + bucket: opts.name, + target_arn: opts.replication_arn, + limit_bytes_per_sec: details.limit_bytes_per_sec, + current_bandwidth_bytes_per_sec: details.current_bandwidth_bytes_per_sec, + }) + .collect(), + ) +} + +pub(crate) async fn obs_ilm_runtime_snapshot() -> ObsIlmRuntimeSnapshot { + let expiry_pending_tasks = { + let expiry = rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState + .read() + .await; + usize_to_u64_saturating(expiry.pending_tasks()) + }; + let transition = &rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_TransitionState; + + ObsIlmRuntimeSnapshot { + expiry_pending_tasks, + transition_active_tasks: i64_to_u64_floor_zero(transition.active_tasks()), + transition_pending_tasks: usize_to_u64_saturating(transition.pending_tasks()), + transition_missed_immediate_tasks: i64_to_u64_floor_zero(transition.missed_immediate_tasks()), + transition_queue_full_tasks: i64_to_u64_floor_zero(transition.queue_full_tasks()), + transition_queue_send_timeout_tasks: i64_to_u64_floor_zero(transition.queue_send_timeout_tasks()), + transition_compensation_scheduled_tasks: i64_to_u64_floor_zero(transition.compensation_scheduled_tasks()), + transition_compensation_running_tasks: i64_to_u64_floor_zero(transition.compensation_running_tasks()), + } +} + +pub(crate) async fn obs_bucket_replication_detail_stats() -> Vec { + let Some(stats) = rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS.get() 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(MetricBucketReplicationStats { + 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_site_replication_stats() -> MetricReplicationStats { + let Some(stats) = rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS.get() else { + return MetricReplicationStats::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::(); + + MetricReplicationStats { + 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), + } +} + +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) } diff --git a/crates/s3select-api/src/object_store.rs b/crates/s3select-api/src/object_store.rs index 779dbb1fb..c4f09a322 100644 --- a/crates/s3select-api/src/object_store.rs +++ b/crates/s3select-api/src/object_store.rs @@ -12,12 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::storage_compat::ecstore::error::{ - StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, +use crate::storage_compat::{ + SELECT_DEFAULT_READ_BUFFER_SIZE, SelectGetObjectReader, SelectObjectInfo, SelectObjectOptions, SelectStorageError, + SelectStore, resolve_select_object_store_handle, select_default_read_buffer_size_u64, select_is_err_bucket_not_found, + select_is_err_object_not_found, select_is_err_version_not_found, }; -use crate::storage_compat::ecstore::resolve_object_store_handle; -use crate::storage_compat::ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; -use crate::storage_compat::ecstore::store::ECStore; use async_trait::async_trait; use bytes::Bytes; use chrono::Utc; @@ -50,8 +49,6 @@ use tokio::io::{AsyncRead, ReadBuf}; use tokio_util::io::ReaderStream; use transform_stream::AsyncTryStream; -use crate::storage_compat::{SelectGetObjectReader, SelectObjectInfo, SelectObjectOptions}; - /// Maximum allowed object size for JSON DOCUMENT mode. /// /// JSON DOCUMENT format requires loading the entire file into memory for DOM @@ -83,7 +80,7 @@ pub struct EcObjectStore { /// this key in the root JSON object before flattening. json_sub_path: Option, - store: Arc, + store: Arc, } #[derive(Clone, Copy, Debug)] @@ -111,7 +108,7 @@ pub struct InvalidScanRange; impl EcObjectStore { pub fn new(input: Arc) -> S3Result { - let Some(store) = resolve_object_store_handle() else { + let Some(store) = resolve_select_object_store_handle() else { return Err(s3_error!(InternalError, "ec store not inited")); }; @@ -236,7 +233,7 @@ impl EcObjectStore { return Ok(Bytes::new()); } - let mut end = (DEFAULT_READ_BUFFER_SIZE as u64).min(object_size); + let mut end = select_default_read_buffer_size_u64().min(object_size); loop { let bytes = self.read_raw_range_with_opts(0..end, opts).await?; if let Some(pos) = find_delimiter(&bytes, delimiter) { @@ -333,8 +330,8 @@ fn find_delimiter(bytes: &[u8], delimiter: &[u8]) -> Option { bytes.windows(delimiter.len()).position(|window| window == delimiter) } -fn map_storage_error(bucket: &str, object: &str, err: StorageError) -> o_Error { - if is_err_bucket_not_found(&err) || is_err_object_not_found(&err) || is_err_version_not_found(&err) { +fn map_storage_error(bucket: &str, object: &str, err: SelectStorageError) -> o_Error { + if select_is_err_bucket_not_found(&err) || select_is_err_object_not_found(&err) || select_is_err_version_not_found(&err) { return o_Error::NotFound { path: format!("{bucket}/{object}"), source: err.to_string().into(), @@ -461,7 +458,7 @@ impl ObjectStore for EcObjectStore { GetResultPayload::Stream(stream::empty().boxed()) } else if options.range.is_some() { let size = (result_range.end - result_range.start) as usize; - let stream = bytes_stream(ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE), size).boxed(); + let stream = bytes_stream(ReaderStream::with_capacity(reader.stream, SELECT_DEFAULT_READ_BUFFER_SIZE), size).boxed(); GetResultPayload::Stream(stream) } else if self.is_json_document { // JSON DOCUMENT mode: gate on object size before doing any I/O. @@ -501,7 +498,7 @@ impl ObjectStore for EcObjectStore { None }; let stream = scan_range_stream( - ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE), + ReaderStream::with_capacity(reader.stream, SELECT_DEFAULT_READ_BUFFER_SIZE), delimiter, scan_range, include_header && header.is_none(), @@ -516,14 +513,17 @@ impl ObjectStore for EcObjectStore { GetResultPayload::Stream(convert_field_delimiter_stream(stream, self.need_convert.then(|| self.delimiter.clone()))) } else if self.need_convert { let stream = bytes_stream( - ReaderStream::with_capacity(ConvertStream::new(reader.stream, self.delimiter.clone()), DEFAULT_READ_BUFFER_SIZE), + ReaderStream::with_capacity( + ConvertStream::new(reader.stream, self.delimiter.clone()), + SELECT_DEFAULT_READ_BUFFER_SIZE, + ), original_size as usize, ) .boxed(); GetResultPayload::Stream(stream) } else { let stream = bytes_stream( - ReaderStream::with_capacity(reader.stream, DEFAULT_READ_BUFFER_SIZE), + ReaderStream::with_capacity(reader.stream, SELECT_DEFAULT_READ_BUFFER_SIZE), original_size as usize, ) .boxed(); @@ -587,7 +587,7 @@ impl ConvertStream { ConvertStream { inner, converter: DelimiterConverter::new(delimiter.into_bytes()), - read_buf: vec![0; DEFAULT_READ_BUFFER_SIZE], + read_buf: vec![0; SELECT_DEFAULT_READ_BUFFER_SIZE], pending: Vec::new(), pending_pos: 0, eof: false, @@ -607,7 +607,7 @@ impl AsyncRead for ConvertStream { return Poll::Ready(Ok(())); } - let read_len = DEFAULT_READ_BUFFER_SIZE.min(buf.remaining().max(1)); + let read_len = SELECT_DEFAULT_READ_BUFFER_SIZE.min(buf.remaining().max(1)); let bytes_read = { let mut read_buf = ReadBuf::new(&mut this.read_buf[..read_len]); ready!(Pin::new(&mut *this.inner).poll_read(cx, &mut read_buf))?; diff --git a/crates/s3select-api/src/storage_compat.rs b/crates/s3select-api/src/storage_compat.rs index b86ed950d..b2878e9dd 100644 --- a/crates/s3select-api/src/storage_compat.rs +++ b/crates/s3select-api/src/storage_compat.rs @@ -12,10 +12,30 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub(crate) mod ecstore { - pub(crate) use rustfs_ecstore::{error, resolve_object_store_handle, set_disk, store}; -} - pub(crate) type SelectGetObjectReader = rustfs_ecstore::store_api::GetObjectReader; pub(crate) type SelectObjectInfo = rustfs_ecstore::store_api::ObjectInfo; pub(crate) type SelectObjectOptions = rustfs_ecstore::store_api::ObjectOptions; +pub(crate) type SelectStorageError = rustfs_ecstore::error::StorageError; +pub(crate) type SelectStore = rustfs_ecstore::store::ECStore; + +pub(crate) const SELECT_DEFAULT_READ_BUFFER_SIZE: usize = rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; + +pub(crate) fn select_default_read_buffer_size_u64() -> u64 { + u64::try_from(SELECT_DEFAULT_READ_BUFFER_SIZE).unwrap_or(u64::MAX) +} + +pub(crate) fn resolve_select_object_store_handle() -> Option> { + rustfs_ecstore::resolve_object_store_handle() +} + +pub(crate) fn select_is_err_bucket_not_found(err: &SelectStorageError) -> bool { + rustfs_ecstore::error::is_err_bucket_not_found(err) +} + +pub(crate) fn select_is_err_object_not_found(err: &SelectStorageError) -> bool { + rustfs_ecstore::error::is_err_object_not_found(err) +} + +pub(crate) fn select_is_err_version_not_found(err: &SelectStorageError) -> bool { + rustfs_ecstore::error::is_err_version_not_found(err) +} diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index ec726a9b0..13a58b9c2 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -1105,6 +1105,82 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: migration guard, formatting check, diff hygiene, risk scan, focused script check, and full pre-commit required before push. +- [x] `API-042` Split notify event object contract from ECStore ObjectInfo. + - Current branch: `overtrue/arch-compat-passthrough-contracts`. + - Current slice: give `rustfs-notify` its own lightweight + `NotifyObjectInfo` event DTO, keep ECStore-to-notify conversion private to + the notify compatibility boundary, and update RustFS event handoff sites to + use the conversion explicitly. + - Acceptance: notify no longer publicly re-exports ECStore `ObjectInfo` as + its event object type; existing RustFS event generation, restore-completed + event data, version IDs, object metadata filtering, and ECStore bridge + behavior are preserved. + - Must preserve: S3 event JSON shape, remove-event metadata suppression, + restore-completed glacier data formatting, object key URL encoding, + request/response headers, replication request filtering, and existing + EventArgsBuilder call sites. + - Risk defense: this is a consumer contract split only; ECStore remains the + producer of storage metadata, while notify owns the event-facing DTO. + - Verification: focused notify/RustFS compile, migration and layer guards, + formatting check, diff hygiene, risk scan, full pre-commit, and required + three-expert review passed. + +- [x] `API-043` Remove notify ECStore config passthroughs. + - Current branch: `overtrue/arch-compat-passthrough-contracts`. + - Current slice: replace notify's public compatibility passthroughs for + ECStore config/global modules with a crate-local config update boundary, + then shrink the passthrough guard snapshot. + - Acceptance: notify config mutation code no longer reaches through + ECStore config/global modules directly; the storage compatibility boundary + owns ECStore handle resolution, read, save, and error classification. + - Must preserve: target config read-modify-save behavior, unchanged-config + no-op handling, storage-not-initialized error wording, read/save error + mapping, target reload ordering, and runtime lifecycle logging. + - Risk defense: this keeps persistence semantics unchanged while reducing + the compatibility surface visible to notify business logic. + - Verification: focused notify/RustFS compile, migration and layer guards, + formatting check, diff hygiene, risk scan, full pre-commit, and required + three-expert review required before push. + +- [x] `API-044` Remove S3 Select ECStore module passthroughs. + - Current branch: `overtrue/arch-compat-passthrough-contracts`. + - Current slice: replace S3 Select's public compatibility passthroughs for + ECStore error, store, set-disk, and resolver modules with crate-local + aliases/functions, then shrink the passthrough guard snapshot. + - Acceptance: S3 Select object-store code no longer reaches through ECStore + modules directly; storage errors, store handle resolution, ECStore store + type ownership, and default read-buffer sizing remain behind the local + storage compatibility boundary. + - Must preserve: S3 Select object-store initialization, not-found error + mapping, scan-range defaults, stream buffer sizing, JSON document handling, + CSV conversion streams, and ECStore object reader/info calls. + - Risk defense: this changes import ownership only; S3 Select still uses the + same ECStore runtime APIs through narrower local compatibility names. + - Verification: focused S3 Select/notify/RustFS compile, migration and layer + guards, formatting check, diff hygiene, risk scan, full pre-commit, and + required three-expert review required before push. + +- [x] `API-045` Remove observability ECStore module passthroughs. + - Current branch: `overtrue/arch-compat-passthrough-contracts`. + - Current slice: replace OBS metrics passthroughs for ECStore bucket, + data-usage, global, pools, and object-store resolver modules with + crate-local storage compatibility functions and snapshots, then shrink the + passthrough guard snapshot. + - Acceptance: OBS metrics collection no longer reaches through ECStore + modules directly; object-store resolution, data-usage loading, capacity + calculation, quota reads, replication state, bucket bandwidth monitor + access, and ILM runtime counters remain behind the OBS compatibility + boundary. + - Must preserve: cluster/health metrics, bucket usage metrics, replication + and bandwidth metrics, scheduler tombstone behavior, disk/drive metrics, + erasure-set metrics, ILM metrics, existing warning paths, and no-data + fallback behavior. + - Risk defense: this changes compatibility ownership only; OBS still reads + the same ECStore runtime state through narrower local compatibility names. + - Verification: focused OBS/notify/S3 Select/RustFS compile, migration and + layer guards, formatting check, diff hygiene, risk scan, full pre-commit, + and required three-expert review required before push. + ## Phase 8 Background Controller Tasks - [x] `BGC-001` Inventory background services. @@ -1381,14 +1457,66 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | API-041 is guard-only and snapshots local ECStore compatibility passthroughs without changing runtime boundaries. | -| Migration preservation | passed | Existing local `storage_compat.rs` paths and ECStore concrete ownership remain unchanged; the guard only reports passthrough drift. | -| Testing/verification | passed | Script syntax, guards, formatting, diff hygiene, risk scan, and full `make pre-commit` passed. | +| Quality/architecture | passed | API-042/API-043/API-044/API-045 narrow notify, S3 Select, and OBS compatibility contracts without moving ECStore storage metadata ownership. | +| Migration preservation | passed | Event builder call sites, ECStore event bridge conversion, restore event data, version IDs, metadata filtering, config read/save semantics, S3 Select store/error/buffer semantics, OBS metrics state reads, unchanged no-op handling, and remove-event behavior are preserved. | +| Testing/verification | passed | Focused compiles/tests, guards, formatting, diff hygiene, risk scan, and full `make pre-commit` passed. | ## Verification Notes Passed before push: +- API-042 current slice: + - `cargo check --tests -p rustfs-notify -p rustfs`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan: passed; no new unwrap/expect, numeric casts, string error + public APIs, boxed public errors, production println/eprintln, or relaxed + ordering introduced in changed Rust files. + - `make pre-commit`: passed. + +- API-043 current slice: + - `cargo test -p rustfs-notify + storage_compat::tests::ecstore_object_info_conversion_preserves_notify_event_fields`: + passed. + - `cargo check --tests -p rustfs-notify -p rustfs`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan: passed; no new unwrap/expect, numeric casts, string error + public APIs, boxed public errors, production println/eprintln, or relaxed + ordering introduced in changed Rust files. + - `make pre-commit`: passed, including 6245 nextest tests passed and 111 + skipped. + +- API-044 current slice: + - `cargo check --tests -p rustfs-s3select-api -p rustfs-notify -p + rustfs`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan: passed; no new unwrap/expect, numeric casts, string error + public APIs, boxed public errors, production println/eprintln, or relaxed + ordering introduced in changed Rust files. + - `make pre-commit`: passed, including 6245 nextest tests passed and 111 + skipped. + +- API-045 current slice: + - `cargo check --tests -p rustfs-obs -p rustfs-s3select-api -p + rustfs-notify -p rustfs`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan: passed; no new unwrap/expect, numeric casts, string error + public APIs, boxed public errors, production println/eprintln, or relaxed + ordering introduced in changed Rust files. + - `make pre-commit`: passed, including 6245 nextest tests passed and 111 + skipped. + - API-041 current slice: - `bash -n scripts/check_architecture_migration_rules.sh`: passed. - `./scripts/check_architecture_migration_rules.sh`: passed. diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 6659b79dd..76fad2925 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -4829,7 +4829,7 @@ impl DefaultObjectUsecase { let event_args = rustfs_notify::EventArgs { event_name: put_event_name_for_post_object(false), bucket_name: bucket.clone(), - object: obj_info.clone(), + object: obj_info.clone().into(), req_params: req_params.clone(), resp_elements: extract_resp_elements(&S3Response::new(output.clone())), version_id: version_id.clone(), diff --git a/rustfs/src/server/event.rs b/rustfs/src/server/event.rs index dbdeda5b4..2654aa4cb 100644 --- a/rustfs/src/server/event.rs +++ b/rustfs/src/server/event.rs @@ -59,7 +59,7 @@ fn convert_ecstore_event_args(args: EcstoreEventArgs) -> Option Some(NotifyEventArgs { event_name, bucket_name: args.bucket_name, - object: args.object, + object: args.object.into(), req_params, resp_elements, version_id, diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 6de9dc167..2d5ab72c0 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -441,21 +441,10 @@ crates/iam/src/storage_compat.rs:error crates/iam/src/storage_compat.rs:global crates/iam/src/storage_compat.rs:notification_sys crates/iam/src/storage_compat.rs:store -crates/notify/src/storage_compat.rs:config -crates/notify/src/storage_compat.rs:global -crates/obs/src/storage_compat.rs:bucket -crates/obs/src/storage_compat.rs:data_usage -crates/obs/src/storage_compat.rs:global -crates/obs/src/storage_compat.rs:pools -crates/obs/src/storage_compat.rs:resolve_object_store_handle crates/protocols/src/swift/storage_compat.rs:bucket crates/protocols/src/swift/storage_compat.rs:error crates/protocols/src/swift/storage_compat.rs:resolve_object_store_handle crates/protocols/src/swift/storage_compat.rs:store -crates/s3select-api/src/storage_compat.rs:error -crates/s3select-api/src/storage_compat.rs:resolve_object_store_handle -crates/s3select-api/src/storage_compat.rs:set_disk -crates/s3select-api/src/storage_compat.rs:store crates/scanner/src/storage_compat.rs:bucket crates/scanner/src/storage_compat.rs:cache_value crates/scanner/src/storage_compat.rs:config