mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-27 15:37:02 +00:00
add versioning_sys
This commit is contained in:
@@ -11,6 +11,7 @@ mod replication;
|
|||||||
pub mod tags;
|
pub mod tags;
|
||||||
mod target;
|
mod target;
|
||||||
pub mod utils;
|
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};
|
pub use metadata_sys::{bucket_metadata_sys_set, get_bucket_metadata_sys, init_bucket_metadata_sys};
|
||||||
|
|||||||
@@ -1,14 +1,20 @@
|
|||||||
use crate::error::Result;
|
use crate::error::{Error, Result};
|
||||||
use rmp_serde::Serializer as rmpSerializer;
|
use rmp_serde::Serializer as rmpSerializer;
|
||||||
use serde::{Deserialize, Serialize};
|
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)]
|
#[derive(Debug, Clone, PartialEq, Eq, Hash, Default, Deserialize, Serialize)]
|
||||||
pub enum State {
|
pub enum State {
|
||||||
#[default]
|
#[default]
|
||||||
Enabled,
|
|
||||||
Suspended,
|
Suspended,
|
||||||
// 如果未来可能会使用到Disabled状态,可以在这里添加
|
Enabled,
|
||||||
// Disabled,
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 实现Display trait用于打印
|
// 实现Display trait用于打印
|
||||||
@@ -20,8 +26,6 @@ impl std::fmt::Display for State {
|
|||||||
match *self {
|
match *self {
|
||||||
State::Enabled => "Enabled",
|
State::Enabled => "Enabled",
|
||||||
State::Suspended => "Suspended",
|
State::Suspended => "Suspended",
|
||||||
// 如果未来可能会使用到Disabled状态,可以在这里添加
|
|
||||||
// State::Disabled => "Disabled",
|
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -52,4 +56,83 @@ impl Versioning {
|
|||||||
let t: Versioning = rmp_serde::from_slice(buf)?;
|
let t: Versioning = rmp_serde::from_slice(buf)?;
|
||||||
Ok(t)
|
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
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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<Versioning> {
|
||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -4,3 +4,4 @@ pub mod fs;
|
|||||||
pub mod hash;
|
pub mod hash;
|
||||||
pub mod net;
|
pub mod net;
|
||||||
pub mod path;
|
pub mod path;
|
||||||
|
mod wildcard;
|
||||||
|
|||||||
@@ -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()
|
||||||
|
}
|
||||||
@@ -2,6 +2,8 @@ use bytes::Bytes;
|
|||||||
use ecstore::bucket::get_bucket_metadata_sys;
|
use ecstore::bucket::get_bucket_metadata_sys;
|
||||||
use ecstore::bucket::metadata::BUCKET_TAGGING_CONFIG;
|
use ecstore::bucket::metadata::BUCKET_TAGGING_CONFIG;
|
||||||
use ecstore::bucket::tags::Tags;
|
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::disk::error::DiskError;
|
||||||
use ecstore::new_object_layer_fn;
|
use ecstore::new_object_layer_fn;
|
||||||
use ecstore::store_api::BucketOptions;
|
use ecstore::store_api::BucketOptions;
|
||||||
@@ -854,16 +856,47 @@ impl S3 for FS {
|
|||||||
#[tracing::instrument(level = "debug", skip(self))]
|
#[tracing::instrument(level = "debug", skip(self))]
|
||||||
async fn get_bucket_versioning(
|
async fn get_bucket_versioning(
|
||||||
&self,
|
&self,
|
||||||
_req: S3Request<GetBucketVersioningInput>,
|
req: S3Request<GetBucketVersioningInput>,
|
||||||
) -> S3Result<S3Response<GetBucketVersioningOutput>> {
|
) -> S3Result<S3Response<GetBucketVersioningOutput>> {
|
||||||
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))]
|
#[tracing::instrument(level = "debug", skip(self))]
|
||||||
async fn put_bucket_versioning(
|
async fn put_bucket_versioning(
|
||||||
&self,
|
&self,
|
||||||
_req: S3Request<PutBucketVersioningInput>,
|
req: S3Request<PutBucketVersioningInput>,
|
||||||
) -> S3Result<S3Response<PutBucketVersioningOutput>> {
|
) -> S3Result<S3Response<PutBucketVersioningOutput>> {
|
||||||
|
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"))
|
Err(s3_error!(NotImplemented, "PutBucketVersioning is not implemented yet"))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user