From 1bbd5168c9d7f2ef57544d4a3de182cc660b473f Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 9 Oct 2024 20:45:34 +0800 Subject: [PATCH] add versioning_sys --- ecstore/src/bucket/mod.rs | 3 +- ecstore/src/bucket/versioning/mod.rs | 95 ++++++++++++++++++++++++++-- ecstore/src/bucket/versioning_sys.rs | 61 ++++++++++++++++++ ecstore/src/utils/mod.rs | 1 + ecstore/src/utils/wildcard.rs | 70 ++++++++++++++++++++ rustfs/src/storage/ecfs.rs | 39 +++++++++++- 6 files changed, 259 insertions(+), 10 deletions(-) create mode 100644 ecstore/src/bucket/versioning_sys.rs create mode 100644 ecstore/src/utils/wildcard.rs diff --git a/ecstore/src/bucket/mod.rs b/ecstore/src/bucket/mod.rs index a905fba13..59e35a47b 100644 --- a/ecstore/src/bucket/mod.rs +++ b/ecstore/src/bucket/mod.rs @@ -11,6 +11,7 @@ mod replication; pub mod tags; mod target; pub mod utils; -mod versioning; +pub mod versioning; +pub mod versioning_sys; pub use metadata_sys::{bucket_metadata_sys_set, get_bucket_metadata_sys, init_bucket_metadata_sys}; diff --git a/ecstore/src/bucket/versioning/mod.rs b/ecstore/src/bucket/versioning/mod.rs index aa2f473c3..0cf37eb81 100644 --- a/ecstore/src/bucket/versioning/mod.rs +++ b/ecstore/src/bucket/versioning/mod.rs @@ -1,14 +1,20 @@ -use crate::error::Result; +use crate::error::{Error, Result}; use rmp_serde::Serializer as rmpSerializer; use serde::{Deserialize, Serialize}; +#[derive(Debug, thiserror::Error)] +pub enum VersioningErr { + #[error("too many excluded prefixes")] + TooManyExcludedPrefixes, + #[error("excluded prefixes extension supported only when versioning is enabled")] + ExcludedPrefixNotSupported, +} + #[derive(Debug, Clone, PartialEq, Eq, Hash, Default, Deserialize, Serialize)] pub enum State { #[default] - Enabled, Suspended, - // 如果未来可能会使用到Disabled状态，可以在这里添加 - // Disabled, + Enabled, } // 实现Display trait用于打印 @@ -20,8 +26,6 @@ impl std::fmt::Display for State { match *self { State::Enabled => "Enabled", State::Suspended => "Suspended", - // 如果未来可能会使用到Disabled状态，可以在这里添加 - // State::Disabled => "Disabled", } ) } @@ -52,4 +56,83 @@ impl Versioning { let t: Versioning = rmp_serde::from_slice(buf)?; Ok(t) } + + pub fn validate(&self) -> Result<()> { + match self.status { + State::Suspended => { + if self.excluded_prefixes.len() > 0 { + return Err(Error::new(VersioningErr::ExcludedPrefixNotSupported)); + } + } + State::Enabled => { + if self.excluded_prefixes.len() > 10 { + return Err(Error::new(VersioningErr::TooManyExcludedPrefixes)); + } + } + _ => return Err(Error::msg(format!("unsupported versioning status {}", self.status))), + } + + Ok(()) + } + + pub fn enabled(&self) -> bool { + self.status == State::Enabled + } + + pub fn versioned(&self, prefix: &str) -> bool { + self.prefix_enabled(prefix) || self.prefix_suspended(prefix) + } + + pub fn prefix_enabled(&self, prefix: &str) -> bool { + if self.status != State::Enabled { + return false; + } + + if prefix.is_empty() { + return true; + } + if self.exclude_folders && prefix.ends_with("/") { + return false; + } + + for sprefix in self.excluded_prefixes.iter() { + let full_prefix = format!("{}*", sprefix.prefix); + if utils::wildcard::match_simple(full_prefix, prefix) { + return false; + } + } + return true; + } + + pub fn suspended(&self) -> bool { + self.status == State::Suspended + } + + pub fn prefix_suspended(&self, prefix: &str) -> bool { + if self.status == State::Suspended { + return true; + } + + if self.status == State::Enabled { + if prefix.is_empty() { + return false; + } + + if self.exclude_folders && prefix.starts_with("/") { + return true; + } + + for sprefix in self.excluded_prefixes.iter() { + let full_prefix = format!("{}*", sprefix.prefix); + if utils::wildcard::match_simple(full_prefix, prefix) { + return true; + } + } + } + return false; + } + + pub fn prefixes_excluded(&self) -> bool { + self.excluded_prefixes.len() > 0 || self.exclude_folders + } } diff --git a/ecstore/src/bucket/versioning_sys.rs b/ecstore/src/bucket/versioning_sys.rs new file mode 100644 index 000000000..636bbd8a3 --- /dev/null +++ b/ecstore/src/bucket/versioning_sys.rs @@ -0,0 +1,61 @@ +use super::get_bucket_metadata_sys; +use super::versioning::Versioning; +use crate::disk::RUSTFS_META_BUCKET; +use tracing::warn; + +pub struct BucketVersioningSys {} + +impl BucketVersioningSys { + pub async fn enabled(bucket: &str) -> bool { + match Self::get(bucket).await { + Ok(res) => res.enabled(), + Err(err) => { + warn!("{:?}", err); + false + } + } + } + + pub async fn prefix_enabled(bucket: &str, prefix: &str) -> bool { + match Self::get(bucket).await { + Ok(res) => res.prefix_enabled(prefix), + Err(err) => { + warn!("{:?}", err); + false + } + } + } + + pub async fn suspended(bucket: &str) -> bool { + match Self::get(bucket).await { + Ok(res) => res.suspended(), + Err(err) => { + warn!("{:?}", err); + false + } + } + } + + pub async fn prefix_suspended(bucket: &str, prefix: &str) -> bool { + match Self::get(bucket).await { + Ok(res) => res.prefix_suspended(prefix), + Err(err) => { + warn!("{:?}", err); + false + } + } + } + + pub async fn get(bucket: &str) -> Result { + if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) { + return Ok(Versioning::default()); + } + + let bucket_meta_sys_lock = get_bucket_metadata_sys().await; + let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; + + let (cfg, _) = bucket_meta_sys.get_versioning_config(bucket).await?; + + Ok(cfg) + } +} diff --git a/ecstore/src/utils/mod.rs b/ecstore/src/utils/mod.rs index 90e120d08..969921fb0 100644 --- a/ecstore/src/utils/mod.rs +++ b/ecstore/src/utils/mod.rs @@ -4,3 +4,4 @@ pub mod fs; pub mod hash; pub mod net; pub mod path; +mod wildcard; diff --git a/ecstore/src/utils/wildcard.rs b/ecstore/src/utils/wildcard.rs new file mode 100644 index 000000000..58d81637a --- /dev/null +++ b/ecstore/src/utils/wildcard.rs @@ -0,0 +1,70 @@ +pub fn match_simple(pattern: &str, name: &str) -> bool { + if pattern.is_empty() { + return name == pattern; + } + if pattern == "*" { + return true; + } + // Do an extended wildcard '*' and '?' match. + deep_match_rune(name, pattern, true) +} + +pub fn match_pattern(pattern: &str, name: &str) -> bool { + if pattern.is_empty() { + return name == pattern; + } + if pattern == "*" { + return true; + } + // Do an extended wildcard '*' and '?' match. + deep_match_rune(name, pattern, false) +} + +fn deep_match_rune(str_: &str, pattern: &str, simple: bool) -> bool { + let (mut str_, mut pattern) = (str_.as_bytes(), pattern.as_bytes()); + while !pattern.is_empty() { + match pattern[0] as char { + '*' => { + return if pattern.len() == 1 { + true + } else if deep_match_rune(&str_[..], &pattern[1..], simple) + || (!str_.is_empty() && deep_match_rune(&str_[1..], pattern, simple)) + { + true + } else { + false + }; + } + '?' => { + if str_.is_empty() { + return simple; + } + } + _ => { + if str_.is_empty() || str_[0] != pattern[0] { + return false; + } + } + } + str_ = &str_[1..]; + pattern = &pattern[1..]; + } + str_.is_empty() && pattern.is_empty() +} + +pub fn match_as_pattern_prefix(pattern: &str, text: &str) -> bool { + let mut i = 0; + while i < text.len() && i < pattern.len() { + match pattern.as_bytes()[i] as char { + '*' => return true, + '?' => i += 1, + _ => { + if pattern.as_bytes()[i] != text.as_bytes()[i] { + return false; + } + } + } + i += 1; + } + text.len() <= pattern.len() +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 2b994017a..25e07d008 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -2,6 +2,8 @@ use bytes::Bytes; use ecstore::bucket::get_bucket_metadata_sys; use ecstore::bucket::metadata::BUCKET_TAGGING_CONFIG; use ecstore::bucket::tags::Tags; +use ecstore::bucket::versioning::State as VersioningState; +use ecstore::bucket::versioning_sys::BucketVersioningSys; use ecstore::disk::error::DiskError; use ecstore::new_object_layer_fn; use ecstore::store_api::BucketOptions; @@ -854,16 +856,47 @@ impl S3 for FS { #[tracing::instrument(level = "debug", skip(self))] async fn get_bucket_versioning( &self, - _req: S3Request, + req: S3Request, ) -> S3Result> { - Err(s3_error!(NotImplemented, "GetBucketVersioning is not implemented yet")) + let GetBucketVersioningInput { bucket, .. } = req; + let layer = new_object_layer_fn(); + let lock = layer.read().await; + let store = match lock.as_ref() { + Some(s) => s, + None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))), + }; + + if let Err(e) = store.get_bucket_info(&input.bucket, &BucketOptions::default()).await { + if DiskError::VolumeNotFound.is(&e) { + return Err(s3_error!(NoSuchBucket)); + } else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("{}", e))); + } + } + + let cfg = try_!(BucketVersioningSys::get(&bucket).await); + + let status = match cfg.status { + VersioningState::Enabled => Some(BucketVersioningStatus::ENABLED), + VersioningState::Suspended => Some(BucketVersioningStatus::SUSPENDED), + }; + + Ok(S3Response::new(GetBucketVersioningOutput { + mfa_delete: None, + status, + })) } #[tracing::instrument(level = "debug", skip(self))] async fn put_bucket_versioning( &self, - _req: S3Request, + req: S3Request, ) -> S3Result> { + let PutBucketVersioningInput { bucket, .. } = req; + + // check site replication enable + // check bucket object lock enable + // check replication suspended Err(s3_error!(NotImplemented, "PutBucketVersioning is not implemented yet")) } }