gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-11 07:36:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

merge versioning, fix bug todo

Browse Source
This commit is contained in:
weisd
2024-11-02 00:21:10 +08:00
parent 28dc7379a6
commit 09ea11c13d
65 changed files with 5187 additions and 1966 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/bitrot.rs
+20 -18
View File
@@ -1,25 +1,24 @@
use std::{
any::Any,
collections::HashMap,
io::{Cursor, Read},
};
use blake2::Blake2b512;
use highway::{HighwayHash, HighwayHasher, Key};
use lazy_static::lazy_static;
use sha2::{digest::core_api::BlockSizeUser, Digest, Sha256};
use tokio::{
spawn,
sync::mpsc::{self, Sender},
task::JoinHandle,
};
use crate::{
disk::{error::DiskError, DiskStore},
erasure::{ReadAt, Write},
error::{Error, Result},
store_api::BitrotAlgorithm,
};
use blake2::Blake2b512;
use blake2::Digest as _;
use highway::{HighwayHash, HighwayHasher, Key};
use lazy_static::lazy_static;
use sha2::{digest::core_api::BlockSizeUser, Digest, Sha256};
use std::{
any::Any,
collections::HashMap,
io::{Cursor, Read},
};
use tokio::{
spawn,
sync::mpsc::{self, Sender},
task::JoinHandle,
};
lazy_static! {
static ref BITROT_ALGORITHMS: HashMap<BitrotAlgorithm, &'static str> = {
@@ -84,7 +83,7 @@ impl Hasher {
match self {
Hasher::SHA256(_) => Sha256::block_size(),
Hasher::HighwayHash256(_) => 64,
Hasher::BLAKE2b512(_) => Blake2b512::block_size(),
Hasher::BLAKE2b512(_) => 64,
}
}
@@ -485,7 +484,10 @@ mod test {
use tempfile::TempDir;
use crate::{
bitrot::{new_bitrot_writer, BITROT_ALGORITHMS}, disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption}, error::{Error, Result}, store_api::BitrotAlgorithm
bitrot::{new_bitrot_writer, BITROT_ALGORITHMS},
disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption},
error::{Error, Result},
store_api::BitrotAlgorithm,
};
use super::{bitrot_writer_sum, new_bitrot_reader};
ecstore/src/bucket/metadata.rs
+17 -5
View File
@@ -1,5 +1,8 @@
use super::policy::bucket_policy::BucketPolicy;
use super::{quota::BucketQuota, target::BucketTargets};
use super::object_lock::ObjectLockApi;
use super::versioning::VersioningApi;
use byteorder::{BigEndian, ByteOrder, LittleEndian};
use rmp_serde::Serializer as rmpSerializer;
use s3s::dto::{
@@ -7,7 +10,6 @@ use s3s::dto::{
ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration,
};
use s3s::xml;
use s3s_policy::model::Policy;
use serde::Serializer;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
@@ -70,7 +72,7 @@ pub struct BucketMetadata {
pub new_field_updated_at: OffsetDateTime,
#[serde(skip)]
pub policy_config: Option<Policy>,
pub policy_config: Option<BucketPolicy>,
#[serde(skip)]
pub notification_config: Option<NotificationConfiguration>,
#[serde(skip)]
@@ -149,9 +151,15 @@ impl BucketMetadata {
format!("{}/{}/{}", BUCKET_META_PREFIX, self.name.as_str(), BUCKET_METADATA_FILE)
}
// fn msg_size(&self) -> usize {
// unimplemented!()
// }
pub fn versioning(&self) -> bool {
self.lock_enabled
|| (self.object_lock_config.as_ref().is_some_and(|v| v.enabled())
|| self.versioning_config.as_ref().is_some_and(|v| v.enabled()))
}
pub fn object_locking(&self) -> bool {
self.lock_enabled || (self.versioning_config.as_ref().is_some_and(|v| v.enabled()))
}
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
let mut buf = Vec::new();
@@ -278,6 +286,10 @@ impl BucketMetadata {
Ok(updated)
}
pub fn set_created(&mut self, created: OffsetDateTime) {
self.created = created
}
pub async fn save(&mut self, api: &ECStore) -> Result<()> {
self.parse_all_configs(api)?;
ecstore/src/bucket/metadata_sys.rs
+14 -2
View File
@@ -15,12 +15,12 @@ use s3s::dto::{
BucketLifecycleConfiguration, NotificationConfiguration, ObjectLockConfiguration, ReplicationConfiguration,
ServerSideEncryptionConfiguration, Tagging, VersioningConfiguration,
};
use s3s_policy::model::Policy;
use time::OffsetDateTime;
use tokio::sync::RwLock;
use tracing::{error, warn};
use super::metadata::{deserialize, load_bucket_metadata, BucketMetadata};
use super::policy::bucket_policy::BucketPolicy;
use super::quota::BucketQuota;
use super::target::BucketTargets;
@@ -44,6 +44,11 @@ pub(crate) async fn set_bucket_metadata(bucket: String, bm: BucketMetadata) {
sys.set(bucket, Arc::new(bm)).await
}
pub(crate) async fn get(bucket: &str) -> Result<Arc<BucketMetadata>> {
let sys = GLOBAL_BucketMetadataSys.write().await;
sys.get(bucket).await
}
pub async fn update(bucket: &str, config_file: &str, data: Vec<u8>) -> Result<OffsetDateTime> {
let bucket_meta_sys_lock = get_bucket_metadata_sys().await;
let mut bucket_meta_sys = bucket_meta_sys_lock.write().await;
@@ -114,6 +119,13 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
bucket_meta_sys.get_config_from_disk(bucket).await
}
pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
let bucket_meta_sys_lock = get_bucket_metadata_sys().await;
let bucket_meta_sys = bucket_meta_sys_lock.read().await;
bucket_meta_sys.created_at(bucket).await
}
#[derive(Debug, Default)]
pub struct BucketMetadataSys {
metadata_map: RwLock<HashMap<String, Arc<BucketMetadata>>>,
@@ -373,7 +385,7 @@ impl BucketMetadataSys {
}
}
pub async fn get_bucket_policy(&self, bucket: &str) -> Result<(Policy, OffsetDateTime)> {
pub async fn get_bucket_policy(&self, bucket: &str) -> Result<(BucketPolicy, OffsetDateTime)> {
let bm = match self.get_config(bucket).await {
Ok((res, _)) => res,
Err(err) => {
ecstore/src/bucket/mod.rs
+3
View File
@@ -1,9 +1,12 @@
pub mod error;
pub mod metadata;
pub mod metadata_sys;
pub mod object_lock;
pub mod policy;
pub mod policy_sys;
mod quota;
pub mod tagging;
mod target;
pub mod utils;
pub mod versioning;
pub mod versioning_sys;
ecstore/src/bucket/object_lock/mod.rs
+13
View File
@@ -0,0 +1,13 @@
use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled};
pub trait ObjectLockApi {
fn enabled(&self) -> bool;
}
impl ObjectLockApi for ObjectLockConfiguration {
fn enabled(&self) -> bool {
self.object_lock_enabled
.as_ref()
.is_some_and(|v| v.as_str() == ObjectLockEnabled::ENABLED)
}
}
ecstore/src/bucket/policy/action.rs
+2 -1
View File
@@ -41,6 +41,7 @@ impl AsRef<HashSet<Action>> for ActionSet {
}
}
// TODO:: 使用字符串
// 定义Action枚举类型
#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize, Default, Hash)]
pub enum Action {
@@ -281,7 +282,7 @@ impl Action {
}
}
fn _from_str(s: &str) -> Option<Self> {
pub fn from_str(s: &str) -> Option<Self> {
match s {
"s3:AbortMultipartUpload" => Some(Action::AbortMultipartUpload),
"s3:CreateBucket" => Some(Action::CreateBucket),
ecstore/src/bucket/policy/condition/function.rs
+4 -1
View File
@@ -243,8 +243,11 @@ impl<'de> Deserialize<'de> for Functions {
{
// Instantiate our Visitor and ask the Deserializer to drive
// it over the input data, resulting in an instance of MyMap.
let _map = deserializer.deserialize_map(MyMapVisitor::new())?;
let map = deserializer.deserialize_map(MyMapVisitor::new())?;
for (key, vals) in map.iter() {
println!("functions key {}, vals {:?}", key, vals);
}
// TODO: FIXME: create functions from name
Ok(Functions(Vec::new()))
ecstore/src/bucket/policy/mod.rs
+97
View File
@@ -1,6 +1,103 @@
// use std::collections::HashMap;
// use action::Action;
// use s3s_policy::model::{Effect, Policy, Principal, PrincipalRule, Statement};
// use serde::{Deserialize, Serialize};
// use tower::ready_cache::cache::Equivalent;
// use crate::utils::wildcard;
pub mod action;
pub mod bucket_policy;
pub mod condition;
pub mod effect;
pub mod principal;
pub mod resource;
// #[derive(Debug, Deserialize, Serialize, Default, Clone)]
// pub struct BucketPolicyArgs {
// pub account_name: String,
// pub groups: Vec<String>,
// pub action: Action,
// pub bucket_name: String,
// pub condition_values: HashMap<String, Vec<String>>,
// pub is_owner: bool,
// pub object_name: String,
// }
// pub trait AllowApi {
// fn is_allowed(&self, args: &BucketPolicyArgs) -> bool;
// }
// pub trait MatchApi {
// fn is_match(&self, found: &str) -> bool;
// }
// impl AllowApi for Policy {
// fn is_allowed(&self, args: &BucketPolicyArgs) -> bool {
// for statement in self.statement.as_slice().iter() {
// if statement.effect == Effect::Deny {
// if !statement.is_allowed(args) {
// return false;
// }
// }
// }
// false
// }
// }
// impl AllowApi for Statement {
// fn is_allowed(&self, args: &BucketPolicyArgs) -> bool {
// let check = || -> bool {
// if let Some(principal) = &self.principal {
// if !principal.is_match(&args.account_name) {
// return false;
// }
// }
// false
// };
// self.effect.is_allowed(check())
// }
// }
// impl MatchApi for PrincipalRule {
// fn is_match(&self, found: &str) -> bool {
// match self {
// PrincipalRule::Principal(principal) => match principal {
// Principal::Wildcard => return true,
// Principal::Map(index_map) => {
// if let Some(keys) = index_map.get("AWS") {
// for key in keys.as_slice() {
// if wildcard::match_simple(key, found) {
// return true;
// }
// }
// }
// return false;
// }
// },
// PrincipalRule::NotPrincipal(principal) => match principal {
// Principal::Wildcard => return true,
// Principal::Map(index_map) => todo!(),
// },
// }
// false
// }
// }
// trait EffectApi {
// fn is_allowed(&self, b: bool) -> bool;
// }
// impl EffectApi for Effect {
// fn is_allowed(&self, b: bool) -> bool {
// if self == &Effect::Allow {
// b
// } else {
// !b
// }
// }
// }
ecstore/src/bucket/policy_sys.rs
+18 -24
View File
@@ -1,23 +1,27 @@
use super::metadata_sys::get_bucket_metadata_sys;
use super::{
error::BucketMetadataError,
metadata_sys::get_bucket_metadata_sys,
policy::bucket_policy::{BucketPolicy, BucketPolicyArgs},
};
use crate::error::Result;
use s3s_policy::model::Policy;
use tracing::warn;
pub struct PolicySys {}
impl PolicySys {
// pub async fn is_allowed(args: &BucketPolicyArgs) -> bool {
// match Self::get(&args.bucket_name).await {
// Ok(cfg) => return cfg.is_allowed(args),
// Err(err) => {
// if !BucketMetadataError::BucketPolicyNotFound.is(&err) {
// warn!("config get err {:?}", err);
// }
// }
// }
pub async fn is_allowed(args: &BucketPolicyArgs) -> bool {
match Self::get(&args.bucket_name).await {
Ok(cfg) => return cfg.is_allowed(args),
Err(err) => {
if !BucketMetadataError::BucketPolicyNotFound.is(&err) {
warn!("config get err {:?}", err);
}
}
}
// args.is_owner
// }
pub async fn get(bucket: &str) -> Result<Policy> {
args.is_owner
}
pub async fn get(bucket: &str) -> Result<BucketPolicy> {
let bucket_meta_sys_lock = get_bucket_metadata_sys().await;
let bucket_meta_sys = bucket_meta_sys_lock.write().await;
@@ -26,13 +30,3 @@ impl PolicySys {
Ok(cfg)
}
}
// trait PolicyApi {
// fn is_allowed(&self) -> bool;
// }
// impl PolicyApi for Policy {
// fn is_allowed(&self) -> bool {
// todo!()
// }
// }
ecstore/src/bucket/tagging/mod.rs
+27
View File
@@ -0,0 +1,27 @@
use s3s::dto::Tag;
use url::form_urlencoded;
pub fn decode_tags(tags: &str) -> Vec<Tag> {
let values = form_urlencoded::parse(tags.as_bytes());
let mut list = Vec::new();
for (k, v) in values {
list.push(Tag {
key: k.to_string(),
value: v.to_string(),
});
}
list
}
pub fn encode_tags(tags: Vec<Tag>) -> String {
let mut encoded = form_urlencoded::Serializer::new(String::new());
for tag in tags.iter() {
encoded.append_pair(tag.key.as_str(), tag.value.as_str());
}
encoded.finish()
}
ecstore/src/bucket/utils.rs
+69 -1
View File
@@ -1,5 +1,73 @@
use crate::disk::RUSTFS_META_BUCKET;
use crate::{disk::RUSTFS_META_BUCKET, error::Error};
pub fn is_meta_bucketname(name: &str) -> bool {
name.starts_with(RUSTFS_META_BUCKET)
}
use regex::Regex;
lazy_static::lazy_static! {
static ref VALID_BUCKET_NAME: Regex = Regex::new(r"^[A-Za-z0-9][A-Za-z0-9\.\-\_\:]{1,61}[A-Za-z0-9]$").unwrap();
static ref VALID_BUCKET_NAME_STRICT: Regex = Regex::new(r"^[a-z0-9][a-z0-9\.\-]{1,61}[a-z0-9]$").unwrap();
static ref IP_ADDRESS: Regex = Regex::new(r"^(\d+\.){3}\d+$").unwrap();
}
pub fn check_bucket_name_common(bucket_name: &str, strict: bool) -> Result<(), Error> {
let bucket_name_trimmed = bucket_name.trim();
if bucket_name_trimmed.is_empty() {
return Err(Error::msg("Bucket name cannot be empty"));
}
if bucket_name_trimmed.len() < 3 {
return Err(Error::msg("Bucket name cannot be shorter than 3 characters"));
}
if bucket_name_trimmed.len() > 63 {
return Err(Error::msg("Bucket name cannot be longer than 63 characters"));
}
if bucket_name_trimmed == "rustfs" {
return Err(Error::msg("Bucket name cannot be rustfs"));
}
if IP_ADDRESS.is_match(bucket_name_trimmed) {
return Err(Error::msg("Bucket name cannot be an IP address"));
}
if bucket_name_trimmed.contains("..") || bucket_name_trimmed.contains(".-") || bucket_name_trimmed.contains("-.") {
return Err(Error::msg("Bucket name contains invalid characters"));
}
if strict {
if !VALID_BUCKET_NAME_STRICT.is_match(bucket_name_trimmed) {
return Err(Error::msg("Bucket name contains invalid characters"));
}
} else {
if !VALID_BUCKET_NAME.is_match(bucket_name_trimmed) {
return Err(Error::msg("Bucket name contains invalid characters"));
}
}
Ok(())
}
pub fn check_valid_bucket_name(bucket_name: &str) -> Result<(), Error> {
check_bucket_name_common(bucket_name, false)
}
pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<(), Error> {
check_bucket_name_common(bucket_name, true)
}
pub fn check_valid_object_name_prefix(object_name: &str) -> Result<(), Error> {
if object_name.len() > 1024 {
return Err(Error::msg("Object name cannot be longer than 1024 characters"));
}
if !object_name.is_ascii() {
return Err(Error::msg("Object name with non-UTF-8 strings are not supported"));
}
Ok(())
}
pub fn check_valid_object_name(object_name: &str) -> Result<(), Error> {
if object_name.trim().is_empty() {
return Err(Error::msg("Object name cannot be empty"));
}
check_valid_object_name_prefix(object_name)
}
ecstore/src/bucket/versioning/mod.rs
+54
View File
@@ -0,0 +1,54 @@
use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
pub trait VersioningApi {
fn enabled(&self) -> bool;
fn prefix_enabled(&self, prefix: &str) -> bool;
fn prefix_suspended(&self, prefix: &str) -> bool;
}
impl VersioningApi for VersioningConfiguration {
fn enabled(&self) -> bool {
self.status
.as_ref()
.is_some_and(|v| v.as_str() == BucketVersioningStatus::ENABLED)
}
fn prefix_enabled(&self, prefix: &str) -> bool {
if !self
.status
.as_ref()
.is_some_and(|v| v.as_str() == BucketVersioningStatus::ENABLED)
{
return false;
}
if prefix.is_empty() {
return true;
}
// TODO: ExcludeFolders
true
}
fn prefix_suspended(&self, prefix: &str) -> bool {
if self
.status
.as_ref()
.is_some_and(|v| v.as_str() == BucketVersioningStatus::SUSPENDED)
{
return true;
}
if let Some(status) = self.status.as_ref() {
if status.as_str() == BucketVersioningStatus::ENABLED {
if prefix.is_empty() {
return false;
}
// TODO: ExcludeFolders
}
}
false
}
}
ecstore/src/bucket/versioning_sys.rs
+20 -33
View File
@@ -1,7 +1,7 @@
use super::metadata_sys::get_bucket_metadata_sys;
use super::{metadata_sys::get_bucket_metadata_sys, versioning::VersioningApi};
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::Result;
use s3s::dto::{BucketVersioningStatus, VersioningConfiguration};
use s3s::dto::VersioningConfiguration;
use tracing::warn;
pub struct BucketVersioningSys {}
@@ -17,15 +17,15 @@ impl BucketVersioningSys {
}
}
// 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 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 {
@@ -37,15 +37,15 @@ impl BucketVersioningSys {
// }
// }
// 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 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<VersioningConfiguration> {
if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
@@ -60,16 +60,3 @@ impl BucketVersioningSys {
Ok(cfg)
}
}
trait VersioningApi {
fn enabled(&self) -> bool;
}
impl VersioningApi for VersioningConfiguration {
fn enabled(&self) -> bool {
self.status
.as_ref()
.map(|v| v.as_str() == BucketVersioningStatus::ENABLED)
.is_some_and(|v| v)
}
}
ecstore/src/chunk_stream.rs
+7 -7
View File
@@ -9,19 +9,19 @@ use transform_stream::AsyncTryStream;
pub type SyncBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + Sync + 'a>>;
pub struct ChunkedStream {
pub struct ChunkedStream<'a> {
/// inner
inner: AsyncTryStream<Bytes, StdError, SyncBoxFuture<'static, Result<(), StdError>>>,
inner: AsyncTryStream<Bytes, StdError, SyncBoxFuture<'a, Result<(), StdError>>>,
remaining_length: usize,
}
impl ChunkedStream {
impl<'a> ChunkedStream<'a> {
pub fn new<S>(body: S, content_length: usize, chunk_size: usize, need_padding: bool) -> Self
where
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static,
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'a,
{
let inner = AsyncTryStream::<_, _, SyncBoxFuture<'static, Result<(), StdError>>>::new(|mut y| {
let inner = AsyncTryStream::<_, _, SyncBoxFuture<'a, Result<(), StdError>>>::new(|mut y| {
#[allow(clippy::shadow_same)] // necessary for `pin_mut!`
Box::pin(async move {
pin_mut!(body);
@@ -97,7 +97,7 @@ impl ChunkedStream {
data_size: usize,
) -> Option<Result<(Vec<Bytes>, Bytes), StdError>>
where
S: Stream<Item = Result<Bytes, StdError>> + Send + 'static,
S: Stream<Item = Result<Bytes, StdError>> + Send,
{
let mut bytes_buffer = Vec::new();
@@ -206,7 +206,7 @@ impl ChunkedStream {
// }
}
impl Stream for ChunkedStream {
impl Stream for ChunkedStream<'_> {
type Item = Result<Bytes, StdError>;
fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
ecstore/src/config/common.rs
+145 -4
View File
@@ -1,13 +1,34 @@
use std::collections::HashSet;
use super::error::ConfigError;
use super::{storageclass, Config, GLOBAL_StorageClass, KVS};
use crate::config::error::is_not_found;
use crate::disk::RUSTFS_META_BUCKET;
use crate::error::{Error, Result};
use crate::store::ECStore;
use crate::store_api::{HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader};
use crate::store_api::{HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader, StorageAPI};
use crate::store_err::is_err_object_not_found;
use crate::utils::path::SLASH_SEPARATOR;
use http::HeaderMap;
use lazy_static::lazy_static;
use s3s::dto::StreamingBlob;
use s3s::Body;
use tracing::error;
use super::error::ConfigError;
const CONFIG_PREFIX: &str = "config";
const CONFIG_FILE: &str = "config.json";
pub const STORAGE_CLASS_SUB_SYS: &str = "storage_class";
pub const DEFAULT_KV_KEY: &str = "_";
lazy_static! {
static ref CONFIG_BUCKET: String = format!("{}{}{}", RUSTFS_META_BUCKET, SLASH_SEPARATOR, CONFIG_PREFIX);
static ref SubSystemsDynamic: HashSet<String> = {
let mut h = HashSet::new();
h.insert(STORAGE_CLASS_SUB_SYS.to_owned());
h
};
}
pub async fn read_config(api: &ECStore, file: &str) -> Result<Vec<u8>> {
let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?;
@@ -17,7 +38,16 @@ pub async fn read_config(api: &ECStore, file: &str) -> Result<Vec<u8>> {
async fn read_config_with_metadata(api: &ECStore, file: &str, opts: &ObjectOptions) -> Result<(Vec<u8>, ObjectInfo)> {
let range = HTTPRangeSpec::nil();
let h = HeaderMap::new();
let mut rd = api.get_object_reader(RUSTFS_META_BUCKET, file, range, h, opts).await?;
let mut rd = api
.get_object_reader(RUSTFS_META_BUCKET, file, range, h, opts)
.await
.map_err(|err| {
if is_err_object_not_found(&err) {
Error::new(ConfigError::NotFound)
} else {
err
}
})?;
let data = rd.read_all().await?;
@@ -46,9 +76,120 @@ async fn save_config_with_opts(api: &ECStore, file: &str, data: &[u8], opts: &Ob
.put_object(
RUSTFS_META_BUCKET,
file,
PutObjReader::new(StreamingBlob::from(Body::from(data.to_vec())), data.len()),
&mut PutObjReader::new(StreamingBlob::from(Body::from(data.to_vec())), data.len()),
opts,
)
.await?;
Ok(())
}
fn new_server_config() -> Config {
Config::new()
}
async fn new_and_save_server_config(api: &ECStore) -> Result<Config> {
let mut cfg = new_server_config();
lookup_configs(&mut cfg, api).await;
save_server_config(api, &cfg).await?;
Ok(cfg)
}
pub async fn read_config_without_migrate(api: &ECStore) -> Result<Config> {
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
let data = match read_config(api, config_file.as_str()).await {
Ok(res) => res,
Err(err) => {
if is_not_found(&err) {
let cfg = new_and_save_server_config(api).await?;
return Ok(cfg);
} else {
return Err(err);
}
}
};
read_server_config(api, data.as_slice()).await
}
async fn read_server_config(api: &ECStore, data: &[u8]) -> Result<Config> {
let cfg = {
if data.is_empty() {
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
let cfg_data = match read_config(api, config_file.as_str()).await {
Ok(res) => res,
Err(err) => {
if is_not_found(&err) {
let cfg = new_and_save_server_config(api).await?;
return Ok(cfg);
} else {
return Err(err);
}
}
};
// TODO: decrypt
Config::unmarshal(cfg_data.as_slice())?
} else {
Config::unmarshal(data)?
}
};
Ok(cfg.merge())
}
async fn save_server_config(api: &ECStore, cfg: &Config) -> Result<()> {
let data = cfg.marshal()?;
let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE);
save_config(api, &config_file, data.as_slice()).await
}
pub async fn lookup_configs(cfg: &mut Config, api: &ECStore) {
// TODO: from etcd
if let Err(err) = apply_dynamic_config(cfg, api).await {
error!("apply_dynamic_config err {:?}", &err);
}
}
async fn apply_dynamic_config(cfg: &mut Config, api: &ECStore) -> Result<()> {
for key in SubSystemsDynamic.iter() {
apply_dynamic_config_for_sub_sys(cfg, api, key).await?;
}
Ok(())
}
async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: &ECStore, subsys: &String) -> Result<()> {
let set_drive_counts = api.set_drive_counts();
match subsys.as_str() {
STORAGE_CLASS_SUB_SYS => {
let kvs = match cfg.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_KV_KEY) {
Some(res) => res,
None => KVS::new(),
};
for (i, count) in set_drive_counts.iter().enumerate() {
match storageclass::lookup_config(&kvs, *count) {
Ok(res) => {
if i == 0 {
if GLOBAL_StorageClass.get().is_none() {
if let Err(r) = GLOBAL_StorageClass.set(res) {
error!("GLOBAL_StorageClass.set failed {:?}", r);
}
}
}
}
Err(err) => {
error!("init storageclass err:{:?}", &err);
break;
}
}
}
}
_ => {}
}
Ok(())
}
ecstore/src/config/mod.rs
+136
View File
@@ -1,2 +1,138 @@
pub mod common;
pub mod error;
pub mod storageclass;
use crate::error::Result;
use crate::store::ECStore;
use common::{lookup_configs, read_config_without_migrate, STORAGE_CLASS_SUB_SYS};
use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::OnceLock;
lazy_static! {
pub static ref GLOBAL_StorageClass: OnceLock<storageclass::Config> = OnceLock::new();
pub static ref DefaultKVS: OnceLock<HashMap<String, KVS>> = OnceLock::new();
pub static ref GLOBAL_ServerConfig: OnceLock<Config> = OnceLock::new();
pub static ref GLOBAL_ConfigSys: ConfigSys = ConfigSys::new();
}
pub struct ConfigSys {}
impl ConfigSys {
pub fn new() -> Self {
Self {}
}
pub async fn init(&self, api: &ECStore) -> Result<()> {
let mut cfg = read_config_without_migrate(api).await?;
lookup_configs(&mut cfg, api).await;
let _ = GLOBAL_ServerConfig.set(cfg);
Ok(())
}
}
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct KV {
pub key: String,
pub value: String,
pub hidden_if_empty: bool,
}
#[derive(Debug, Deserialize, Serialize, Clone)]
pub struct KVS(Vec<KV>);
impl KVS {
pub fn new() -> Self {
KVS(Vec::new())
}
pub fn get(&self, key: &str) -> String {
if let Some(v) = self.lookup(key) {
v
} else {
"".to_owned()
}
}
pub fn lookup(&self, key: &str) -> Option<String> {
for kv in self.0.iter() {
if kv.key.as_str() == key {
return Some(kv.value.clone());
}
}
None
}
}
#[derive(Debug, Clone)]
pub struct Config(HashMap<String, HashMap<String, KVS>>);
impl Config {
pub fn new() -> Self {
let mut cfg = Config(HashMap::new());
cfg.set_defaults();
cfg
}
pub fn get_value(&self, subsys: &str, key: &str) -> Option<KVS> {
if let Some(m) = self.0.get(subsys) {
m.get(key).cloned()
} else {
None
}
}
pub fn set_defaults(&mut self) {
if let Some(defaults) = DefaultKVS.get() {
for (k, v) in defaults.iter() {
if !self.0.contains_key(k) {
let mut default = HashMap::new();
default.insert("_".to_owned(), v.clone());
self.0.insert(k.clone(), default);
} else {
if !self.0[k].contains_key("_") {
if let Some(m) = self.0.get_mut(k) {
m.insert("_".to_owned(), v.clone());
}
}
}
}
}
}
pub fn unmarshal(data: &[u8]) -> Result<Config> {
let m: HashMap<String, HashMap<String, KVS>> = serde_json::from_slice(data)?;
let mut cfg = Config(m);
cfg.set_defaults();
Ok(cfg)
}
pub fn marshal(&self) -> Result<Vec<u8>> {
let data = serde_json::to_vec(&self.0)?;
Ok(data)
}
pub fn merge(&self) -> Config {
// TODO: merge defauls
self.clone()
}
}
pub fn register_default_kvs(kvs: HashMap<String, KVS>) {
let mut p = HashMap::new();
for (k, v) in kvs {
p.insert(k, v);
}
let _ = DefaultKVS.set(p);
}
pub fn init() {
let mut kvs = HashMap::new();
kvs.insert(STORAGE_CLASS_SUB_SYS.to_owned(), storageclass::DefaultKVS.clone());
// TODO: other defauls
register_default_kvs(kvs)
}
ecstore/src/config/storageclass.rs
+316
View File
@@ -0,0 +1,316 @@
use std::env;
use crate::{
config::KV,
error::{Error, Result},
};
use super::KVS;
use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use tracing::warn;
// default_partiy_count 默认配置，根据磁盘总数分配校验磁盘数量
pub fn default_partiy_count(drive: usize) -> usize {
match drive {
1 => 0,
2 | 3 => 1,
4 | 5 => 2,
6 | 7 => 3,
_ => 4,
}
}
// Standard constants for all storage class
pub const RRS: &str = "REDUCED_REDUNDANCY";
pub const STANDARD: &str = "STANDARD";
// Standard constants for config info storage class
pub const CLASS_STANDARD: &str = "standard";
pub const CLASS_RRS: &str = "rrs";
pub const OPTIMIZE: &str = "optimize";
pub const INLINE_BLOCK: &str = "inline_block";
// Reduced redundancy storage class environment variable
pub const RRS_ENV: &str = "RUSTFS_STORAGE_CLASS_RRS";
// Standard storage class environment variable
pub const STANDARD_ENV: &str = "RUSTFS_STORAGE_CLASS_STANDARD";
// Optimize storage class environment variable
pub const OPTIMIZE_ENV: &str = "RUSTFS_STORAGE_CLASS_OPTIMIZE";
// Inline block indicates the size of the shard that is considered for inlining
pub const INLINE_BLOCK_ENV: &str = "RUSTFS_STORAGE_CLASS_INLINE_BLOCK";
// Supported storage class scheme is EC
pub const SCHEME_PREFIX: &str = "EC";
// Min parity drives
pub const MIN_PARITY_DRIVES: usize = 0;
// Default RRS parity is always minimum parity.
pub const DEFAULT_RRS_PARITY: usize = 1;
pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024;
lazy_static! {
pub static ref DefaultKVS: KVS = {
let mut kvs = Vec::new();
kvs.push(KV {
key: CLASS_STANDARD.to_owned(),
value: "".to_owned(),
hidden_if_empty: false,
});
kvs.push(KV {
key: CLASS_RRS.to_owned(),
value: "EC:1".to_owned(),
hidden_if_empty: false,
});
kvs.push(KV {
key: OPTIMIZE.to_owned(),
value: "availability".to_owned(),
hidden_if_empty: false,
});
kvs.push(KV {
key: INLINE_BLOCK.to_owned(),
value: "".to_owned(),
hidden_if_empty: true,
});
KVS(kvs)
};
}
// StorageClass - holds storage class information
#[derive(Serialize, Deserialize, Debug, Default)]
pub struct StorageClass {
parity: usize,
}
// Config storage class configuration
#[derive(Serialize, Deserialize, Debug, Default)]
pub struct Config {
standard: StorageClass,
rrs: StorageClass,
optimize: Option<String>,
inline_block: usize,
initialized: bool,
}
impl Config {
pub fn get_parity_for_sc(&self, sc: &str) -> Option<usize> {
match sc.trim() {
RRS => {
if self.initialized {
Some(self.rrs.parity)
} else {
None
}
}
_ => {
if self.initialized {
Some(self.standard.parity)
} else {
None
}
}
}
}
pub fn should_inline(&self, shard_size: usize, versioned: bool) -> bool {
let mut inline_block = DEFAULT_INLINE_BLOCK;
if self.initialized {
inline_block = self.inline_block;
}
if versioned {
shard_size <= inline_block / 8
} else {
shard_size <= inline_block
}
}
pub fn inline_block(&self) -> usize {
if !self.initialized {
DEFAULT_INLINE_BLOCK
} else {
self.inline_block
}
}
pub fn capacity_optimized(&self) -> bool {
if !self.initialized {
false
} else {
self.optimize.as_ref().is_some_and(|v| v.as_str() == "capacity")
}
}
}
pub fn lookup_config(kvs: &KVS, set_drive_count: usize) -> Result<Config> {
let standard = {
let ssc_str = {
if let Ok(ssc_str) = env::var(STANDARD_ENV) {
ssc_str
} else {
kvs.get(CLASS_STANDARD)
}
};
if !ssc_str.is_empty() {
parse_storage_class(&ssc_str)?
} else {
StorageClass {
parity: default_partiy_count(set_drive_count),
}
}
};
let rrs = {
let ssc_str = {
if let Ok(ssc_str) = env::var(RRS_ENV) {
ssc_str
} else {
kvs.get(RRS)
}
};
if !ssc_str.is_empty() {
parse_storage_class(&ssc_str)?
} else {
StorageClass {
parity: {
if set_drive_count == 1 {
0
} else {
DEFAULT_RRS_PARITY
}
},
}
}
};
validate_parity_inner(standard.parity, rrs.parity, set_drive_count)?;
let optimize = {
if let Ok(ev) = env::var(OPTIMIZE_ENV) {
Some(ev)
} else {
None
}
};
let inline_block = {
if let Ok(ev) = env::var(INLINE_BLOCK_ENV) {
if let Ok(block) = ev.parse::<bytesize::ByteSize>() {
if block.as_u64() as usize > DEFAULT_INLINE_BLOCK {
warn!("inline block value bigger than recommended max of 128KiB -> {}, performance may degrade for PUT please benchmark the changes",block);
}
block.as_u64() as usize
} else {
return Err(Error::msg(format!("parse {} format failed", INLINE_BLOCK_ENV)));
}
} else {
DEFAULT_INLINE_BLOCK
}
};
Ok(Config {
standard,
rrs,
optimize,
inline_block,
initialized: true,
})
}
pub fn parse_storage_class(env: &str) -> Result<StorageClass> {
let s: Vec<&str> = env.split(':').collect();
// only two elements allowed in the string - "scheme" and "number of parity drives"
if s.len() != 2 {
return Err(Error::msg(&format!(
"Invalid storage class format: {}. Expected 'Scheme:Number of parity drives'.",
env
)));
}
// only allowed scheme is "EC"
if s[0] != SCHEME_PREFIX {
return Err(Error::msg(&format!("Unsupported scheme {}. Supported scheme is EC.", s[0])));
}
// Number of parity drives should be integer
let parity_drives: usize = match s[1].parse() {
Ok(num) => num,
Err(_) => return Err(Error::msg(&format!("Failed to parse parity value: {}.", s[1]))),
};
Ok(StorageClass { parity: parity_drives })
}
// ValidateParity validates standard storage class parity.
pub fn validate_parity(ss_parity: usize, set_drive_count: usize) -> Result<()> {
if ss_parity > 0 && ss_parity < MIN_PARITY_DRIVES {
return Err(Error::msg(format!(
"parity {} should be greater than or equal to {}",
ss_parity, MIN_PARITY_DRIVES
)));
}
if ss_parity > set_drive_count / 2 {
return Err(Error::msg(format!(
"parity {} should be less than or equal to {}",
ss_parity,
set_drive_count / 2
)));
}
Ok(())
}
// Validates the parity drives.
pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_count: usize) -> Result<()> {
if ss_parity > 0 && ss_parity < MIN_PARITY_DRIVES {
return Err(Error::msg(format!(
"Standard storage class parity {} should be greater than or equal to {}",
ss_parity, MIN_PARITY_DRIVES
)));
}
// RRS parity drives should be greater than or equal to minParityDrives.
// Parity below minParityDrives is not supported.
if rrs_parity > 0 && rrs_parity < MIN_PARITY_DRIVES {
return Err(Error::msg(format!(
"Reduced redundancy storage class parity {} should be greater than or equal to {}",
rrs_parity, MIN_PARITY_DRIVES
)));
}
if set_drive_count > 2 {
if ss_parity > set_drive_count / 2 {
return Err(Error::msg(format!(
"Standard storage class parity {} should be less than or equal to {}",
ss_parity,
set_drive_count / 2
)));
}
if rrs_parity > set_drive_count / 2 {
return Err(Error::msg(format!(
"Reduced redundancy storage class parity {} should be less than or equal to {}",
rrs_parity,
set_drive_count / 2
)));
}
}
if ss_parity > 0 && rrs_parity > 0 {
if ss_parity < rrs_parity {
return Err(Error::msg(format!("Standard storage class parity drives {} should be greater than or equal to Reduced redundancy storage class parity drives {}", ss_parity, rrs_parity)));
}
}
Ok(())
}
ecstore/src/disk/error.rs
+11
View File
@@ -260,6 +260,17 @@ pub fn os_err_to_file_err(e: io::Error) -> Error {
}
}
pub fn is_err_file_not_found(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<DiskError>() {
match e {
DiskError::FileNotFound => true,
_ => false,
}
} else {
false
}
}
pub fn is_sys_err_no_space(e: &io::Error) -> bool {
if let Some(no) = e.raw_os_error() {
return no == 28;
ecstore/src/disk/local.rs
+47 -28
View File
@@ -1,4 +1,6 @@
use super::error::{is_sys_err_io, is_sys_err_not_empty, is_sys_err_too_many_files, os_is_not_exist, os_is_permission};
use super::error::{
is_err_file_not_found, is_sys_err_io, is_sys_err_not_empty, is_sys_err_too_many_files, os_is_not_exist, os_is_permission,
};
use super::os::is_root_disk;
use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
use super::{
@@ -16,17 +18,21 @@ use crate::disk::os::check_path_length;
use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
use crate::error::{Error, Result};
use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
use crate::set_disk::{conv_part_err_to_int, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN, CHECK_PART_VOLUME_NOT_FOUND};
use crate::set_disk::{
conv_part_err_to_int, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
CHECK_PART_VOLUME_NOT_FOUND,
};
use crate::store_api::BitrotAlgorithm;
use crate::utils::fs::{access, lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY};
use crate::utils::os::get_info;
use crate::utils::path::{clean, has_suffix, SLASH_SEPARATOR};
use crate::utils::path::{clean, has_suffix, GLOBAL_DIR_SUFFIX_WITH_SLASH, SLASH_SEPARATOR};
use crate::{
file_meta::FileMeta,
store_api::{FileInfo, RawFileInfo},
utils,
};
use path_absolutize::Absolutize;
use std::collections::HashSet;
use std::fmt::Debug;
use std::io::Cursor;
use std::os::unix::fs::MetadataExt;
@@ -411,7 +417,7 @@ impl LocalDisk {
async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
let mut f = utils::fs::open_file(file_path, O_RDONLY).await?;
let mut f = utils::fs::open_file(file_path.as_ref(), O_RDONLY).await?;
let meta = f.metadata().await?;
@@ -682,7 +688,7 @@ pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Vec<u8>, Option
let (data, meta) = match read_file_all(&p).await {
Ok((data, meta)) => (data, Some(meta)),
Err(e) => {
if DiskError::FileNotFound.is(&e) {
if is_err_file_not_found(&e) {
(Vec::new(), None)
} else {
return Err(e);
@@ -939,7 +945,7 @@ impl DiskAPI for LocalDisk {
}
resp.results[i] = CHECK_PART_SUCCESS;
},
}
Err(err) => {
match os_err_to_file_err(err).downcast_ref() {
Some(DiskError::FileNotFound) => {
@@ -949,20 +955,20 @@ impl DiskAPI for LocalDisk {
ErrorKind::NotFound => {
resp.results[i] = CHECK_PART_VOLUME_NOT_FOUND;
continue;
},
_ => {},
}
_ => {}
}
}
}
resp.results[i] = CHECK_PART_FILE_NOT_FOUND;
},
_ => {},
}
_ => {}
}
continue;
}
}
}
Ok(resp)
}
@@ -1229,7 +1235,7 @@ impl DiskAPI for LocalDisk {
let entries = match os::read_dir(&dir_path_abs, count).await {
Ok(res) => res,
Err(e) => {
if DiskError::FileNotFound.is(&e) && !skip_access_checks(volume) {
if is_err_file_not_found(&e) && !skip_access_checks(volume) {
if let Err(e) = utils::fs::access(&volume_dir).await {
return Err(convert_access_error(e, DiskError::VolumeAccessDenied));
}
@@ -1246,12 +1252,12 @@ impl DiskAPI for LocalDisk {
let mut entries = match self.list_dir("", &opts.bucket, &opts.base_dir, -1).await {
Ok(res) => res,
Err(e) => {
if !DiskError::VolumeNotFound.is(&e) && !DiskError::FileNotFound.is(&e) {
if !DiskError::VolumeNotFound.is(&e) && !is_err_file_not_found(&e) {
error!("list_dir err {:?}", &e);
}
if opts.report_notfound && DiskError::FileNotFound.is(&e) {
return Err(Error::new(DiskError::FileNotFound));
if opts.report_notfound && is_err_file_not_found(&e) {
return Err(e);
}
return Ok(Vec::new());
}
@@ -1270,6 +1276,8 @@ impl DiskAPI for LocalDisk {
let mut metas = Vec::new();
let mut dir_objes = HashSet::new();
// 第一层过滤
for entry in entries.iter() {
// check limit
@@ -1283,14 +1291,24 @@ impl DiskAPI for LocalDisk {
// warn!("walk_dir entry {}", entry);
let mut meta = MetaCacheEntry {
name: entry.clone(),
..Default::default()
};
let mut meta = MetaCacheEntry { ..Default::default() };
let fpath = self.get_object_path(bucket, format!("{}/{}", &meta.name, STORAGE_FORMAT_FILE).as_str())?;
let fpath = self.get_object_path(bucket, format!("{}/{}", &entry, STORAGE_FORMAT_FILE).as_str())?;
meta.metadata = self.read_metadata(&fpath).await.unwrap_or_default();
if let Ok(data) = self.read_metadata(&fpath).await {
meta.metadata = data;
}
let mut name = entry.clone();
if name.ends_with(SLASH_SEPARATOR) {
if name.ends_with(GLOBAL_DIR_SUFFIX_WITH_SLASH) {
name = format!("{}{}", name.as_str().trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH), SLASH_SEPARATOR);
dir_objes.insert(name.clone());
} else {
name = name.as_str().trim_end_matches(SLASH_SEPARATOR).to_owned();
}
}
meta.name = name;
metas.push(meta);
}
@@ -1535,7 +1553,7 @@ impl DiskAPI for LocalDisk {
let mut volumes = Vec::new();
let entries = os::read_dir(&self.root, -1).await.map_err(|e| {
if DiskError::FileAccessDenied.is(&e) || DiskError::FileNotFound.is(&e) {
if DiskError::FileAccessDenied.is(&e) || is_err_file_not_found(&e) {
Error::new(DiskError::DiskAccessDenied)
} else {
e
@@ -1543,7 +1561,7 @@ impl DiskAPI for LocalDisk {
})?;
for entry in entries {
if !utils::path::has_suffix(&entry, SLASH_SEPARATOR) || !Self::is_valid_volname(&entry) {
if !utils::path::has_suffix(&entry, SLASH_SEPARATOR) || !Self::is_valid_volname(utils::path::clean(&entry).as_str()) {
continue;
}
@@ -1600,7 +1618,7 @@ impl DiskAPI for LocalDisk {
Ok(())
}
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) -> Result<()> {
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
if fi.metadata.is_some() {
let volume_dir = self.get_bucket_path(volume)?;
let file_path = volume_dir.join(Path::new(&path));
@@ -1611,7 +1629,7 @@ impl DiskAPI for LocalDisk {
.read_all(volume, format!("{}/{}", &path, super::STORAGE_FORMAT_FILE).as_str())
.await
.map_err(|e| {
if DiskError::FileNotFound.is(&e) && fi.version_id.is_some() {
if is_err_file_not_found(&e) && fi.version_id.is_some() {
Error::new(DiskError::FileVersionNotFound)
} else {
e
@@ -1665,6 +1683,7 @@ impl DiskAPI for LocalDisk {
return Ok(());
}
#[tracing::instrument(level = "debug", skip(self))]
async fn read_version(
&self,
_org_volume: &str,
@@ -1683,7 +1702,7 @@ impl DiskAPI for LocalDisk {
let mut meta = FileMeta::default();
meta.unmarshal_msg(&data)?;
let fi = meta.into_fileinfo(volume, path, version_id, false, true)?;
let fi = meta.into_fileinfo(volume, path, version_id, read_data, true)?;
Ok(fi)
}
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
@@ -1770,7 +1789,7 @@ impl DiskAPI for LocalDisk {
}
}
Err(e) => {
if !(DiskError::FileNotFound.is(&e) || DiskError::VolumeNotFound.is(&e)) {
if !(is_err_file_not_found(&e) || DiskError::VolumeNotFound.is(&e)) {
res.exists = true;
res.error = e.to_string();
}
@@ -1827,7 +1846,7 @@ async fn get_disk_info(drive_path: PathBuf) -> Result<(Info, bool)> {
let drive_path = drive_path.to_string_lossy().to_string();
check_path_length(&drive_path)?;
let disk_info = get_info(&drive_path, false)?;
let disk_info = get_info(&drive_path)?;
let root_drive = if !*GLOBAL_IsErasureSD.read().await {
let root_disk_threshold = *GLOBAL_RootDiskThreshold.read().await;
if root_disk_threshold > 0 {
ecstore/src/disk/mod.rs
+14 -4
View File
@@ -19,7 +19,6 @@ use crate::{
file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion},
store_api::{FileInfo, RawFileInfo},
};
use endpoint::Endpoint;
use futures::StreamExt;
use protos::proto_gen::node_service::{
@@ -35,6 +34,7 @@ use tokio::{
};
use tokio_stream::wrappers::ReceiverStream;
use tonic::{service::interceptor::InterceptedService, transport::Channel, Request, Status, Streaming};
use tracing::error;
use tracing::info;
use uuid::Uuid;
@@ -96,7 +96,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
) -> Result<Vec<Option<Error>>>;
async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()>;
async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>;
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) -> Result<()>;
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()>;
async fn read_version(
&self,
org_volume: &str,
@@ -143,7 +143,7 @@ pub struct CheckPartsResp {
pub results: Vec<usize>,
}
#[derive(Debug, Serialize, Deserialize)]
#[derive(Debug, Serialize, Deserialize, Default)]
pub struct UpdateMetadataOpts {
pub no_persistence: bool,
}
@@ -597,7 +597,7 @@ pub struct VolumeInfo {
pub created: Option<OffsetDateTime>,
}
#[derive(Deserialize, Serialize)]
#[derive(Deserialize, Serialize, Debug)]
pub struct ReadOptions {
pub read_data: bool,
pub healing: bool,
@@ -644,6 +644,7 @@ pub struct ReadOptions {
pub enum FileWriter {
Local(LocalFileWriter),
Remote(RemoteFileWriter),
Buffer(Vec<u8>),
}
#[async_trait::async_trait]
@@ -656,6 +657,10 @@ impl Write for FileWriter {
match self {
Self::Local(local_file_writer) => local_file_writer.write(buf).await,
Self::Remote(remote_file_writer) => remote_file_writer.write(buf).await,
Self::Buffer(buffer) => {
buffer.extend_from_slice(buf);
Ok(())
}
}
}
}
@@ -773,6 +778,7 @@ impl Write for RemoteFileWriter {
pub enum FileReader {
Local(LocalFileReader),
Remote(RemoteFileReader),
Buffer(Vec<u8>),
}
#[async_trait::async_trait]
@@ -781,6 +787,10 @@ impl ReadAt for FileReader {
match self {
Self::Local(local_file_writer) => local_file_writer.read_at(offset, length).await,
Self::Remote(remote_file_writer) => remote_file_writer.read_at(offset, length).await,
Self::Buffer(buffer) => {
let s = &buffer[offset..offset + length];
Ok((s.to_vec(), s.len()))
}
}
}
}
ecstore/src/disk/os.rs
+21 -18
View File
@@ -56,7 +56,7 @@ pub fn is_root_disk(disk_path: &str, root_disk: &str) -> Result<bool> {
return Ok(false);
}
same_disk(disk_path, root_disk)
Ok(same_disk(disk_path, root_disk)?)
}
pub async fn make_dir_all(path: impl AsRef<Path>, base_dir: impl AsRef<Path>) -> Result<()> {
@@ -90,13 +90,14 @@ pub async fn read_dir(path: impl AsRef<Path>, count: i32) -> Result<Vec<String>>
let file_type = entry.file_type().await?;
if file_type.is_dir() {
count -= 1;
if file_type.is_file() {
volumes.push(name);
} else if file_type.is_dir() {
volumes.push(format!("{}{}", name, utils::path::SLASH_SEPARATOR));
if count == 0 {
break;
}
}
count -= 1;
if count == 0 {
break;
}
}
@@ -108,17 +109,19 @@ pub async fn rename_all(
dst_file_path: impl AsRef<Path>,
base_dir: impl AsRef<Path>,
) -> Result<()> {
reliable_rename(src_file_path, dst_file_path, base_dir).await.map_err(|e| {
if is_sys_err_not_dir(&e) || !os_is_not_exist(&e) || is_sys_err_path_not_found(&e) {
Error::new(DiskError::FileAccessDenied)
} else if os_is_not_exist(&e) {
Error::new(DiskError::FileNotFound)
} else if os_is_exist(&e) {
Error::new(DiskError::IsNotRegular)
} else {
Error::new(e)
}
})?;
reliable_rename(src_file_path, dst_file_path.as_ref(), base_dir)
.await
.map_err(|e| {
if is_sys_err_not_dir(&e) || !os_is_not_exist(&e) || is_sys_err_path_not_found(&e) {
Error::new(DiskError::FileAccessDenied)
} else if os_is_not_exist(&e) {
Error::new(DiskError::FileNotFound)
} else if os_is_exist(&e) {
Error::new(DiskError::IsNotRegular)
} else {
Error::new(e)
}
})?;
Ok(())
}
ecstore/src/disk/remote.rs
+9 -4
View File
@@ -4,7 +4,10 @@ use futures::lock::Mutex;
use protos::{
node_service_time_out_client,
proto_gen::node_service::{
CheckPartsRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest
CheckPartsRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest,
DiskInfoRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest,
ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest,
UpdateMetadataRequest, VerifyFileRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
},
};
use tonic::Request;
@@ -12,7 +15,9 @@ use tracing::info;
use uuid::Uuid;
use super::{
endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions
endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader,
RemoteFileWriter, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
};
use crate::{
disk::error::DiskError,
@@ -110,7 +115,7 @@ impl DiskAPI for RemoteDisk {
info!("read_all success");
if !response.success {
return Err(DiskError::FileNotFound.into());
return Err(Error::new(DiskError::FileNotFound));
}
Ok(response.data)
@@ -479,7 +484,7 @@ impl DiskAPI for RemoteDisk {
Ok(())
}
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: UpdateMetadataOpts) -> Result<()> {
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
info!("update_metadata");
let file_info = serde_json::to_string(&fi)?;
let opts = serde_json::to_string(&opts)?;
ecstore/src/disks_layout.rs
+2 -2
View File
@@ -411,8 +411,8 @@ fn get_set_indexes<T: AsRef<str>>(
if !has_set_drive_count {
return Err(Error::from_string(format!(
"Invalid set drive count. Acceptable values for {:?} number drives are {:?}",
common_size, &set_counts
"Invalid set drive count {}. Acceptable values for {:?} number drives are {:?}",
set_drive_count, common_size, &set_counts
)));
}
set_drive_count
ecstore/src/erasure.rs
+57 -50
View File
@@ -3,7 +3,7 @@ use crate::error::{Error, Result, StdError};
use crate::quorum::{object_op_ignored_errs, reduce_write_quorum_errs};
use bytes::Bytes;
use futures::future::join_all;
use futures::{Stream, StreamExt};
use futures::{pin_mut, Stream, StreamExt};
use reed_solomon_erasure::galois_8::ReedSolomon;
use std::any::Any;
use std::fmt::Debug;
@@ -14,7 +14,8 @@ use tracing::warn;
// use tracing::debug;
use uuid::Uuid;
use crate::chunk_stream::ChunkedStream;
use reader::reader::ChunkedStream;
// use crate::chunk_stream::ChunkedStream;
use crate::disk::error::DiskError;
pub struct Erasure {
@@ -45,6 +46,7 @@ impl Erasure {
}
}
#[tracing::instrument(level = "debug", skip(self, body,writers))]
pub async fn encode<S>(
&self,
body: S,
@@ -54,73 +56,78 @@ impl Erasure {
write_quorum: usize,
) -> Result<usize>
where
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static,
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync,
{
let mut stream = ChunkedStream::new(body, total_size, self.block_size, false);
let stream = ChunkedStream::new(body, self.block_size);
// let mut stream = ChunkedStream::new(body, total_size, self.block_size, false);
let mut total: usize = 0;
// let mut idx = 0;
while let Some(result) = stream.next().await {
match result {
Ok(data) => {
total += data.len();
pin_mut!(stream);
// EOF
if data.is_empty() {
break;
}
// warn!("encode start...");
// idx += 1;
// debug!("encode {} get data {}", idx, data.len());
loop {
match stream.next().await {
Some(result) => match result {
Ok(data) => {
total += data.len();
let blocks = self.encode_data(data.as_ref())?;
// EOF
if data.is_empty() {
break;
}
// debug!(
// "encode shard {} size: {}/{} from block_size {}, total_size {} ",
// idx,
// blocks[0].len(),
// blocks.len(),
// data.len(),
// total_size
// );
// idx += 1;
// warn!("encode {} get data {:?}", data.len(), data.to_vec());
let mut errs = Vec::new();
let blocks = self.encode_data(data.as_ref())?;
for (i, w) in writers.iter_mut().enumerate() {
if w.is_none() {
// debug!(
// "encode shard {} size: {}/{} from block_size {}, total_size {} ",
// idx,
// blocks[0].len(),
// blocks.len(),
// data.len(),
// total_size
// );
let mut errs = Vec::new();
for (i, w) in writers.iter_mut().enumerate() {
if w.is_none() {
continue;
}
match w.as_mut().unwrap().write(blocks[i].as_ref()).await {
Ok(_) => errs.push(None),
Err(e) => errs.push(Some(e)),
}
}
let none_count = errs.iter().filter(|&x| x.is_none()).count();
if none_count >= write_quorum {
continue;
}
match w.as_mut().unwrap().write(blocks[i].as_ref()).await {
Ok(_) => errs.push(None),
Err(e) => errs.push(Some(e)),
if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
warn!("Erasure encode errs {:?}", &errs);
return Err(err);
}
}
let none_count = errs.iter().filter(|&x| x.is_none()).count();
if none_count >= write_quorum {
continue;
}
if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
warn!("Erasure encode errs {:?}", &errs);
return Err(err);
Err(e) => {
warn!("poll result err {:?}", &e);
return Err(Error::msg(e.to_string()));
}
},
None => {
// warn!("poll empty result");
break;
}
Err(e) => return Err(Error::from_std_error(e)),
}
}
// debug!(" encode_data done shard block num {}", idx);
let _ = close_bitrot_writers(writers).await?;
Ok(total)
// loop {
// match rd.next().await {
// Some(res) => todo!(),
// None => todo!(),
// }
// }
}
pub async fn decode(
@@ -308,7 +315,7 @@ impl Erasure {
(data_size + self.data_shards - 1) / self.data_shards
}
// returns final erasure size from original size.
fn shard_file_size(&self, total_size: usize) -> usize {
pub fn shard_file_size(&self, total_size: usize) -> usize {
if total_size == 0 {
return 0;
}
@@ -355,7 +362,7 @@ pub trait ReadAt: Debug {
#[derive(Debug)]
pub struct ShardReader {
readers: Vec<Option<BitrotReader>>, // 磁盘
data_block_count: usize, // 总的分片数量
data_block_count: usize, // 总的分片数量
parity_block_count: usize,
shard_size: usize, // 每个分片的块大小 一次读取一块
shard_file_size: usize, // 分片文件总长度
ecstore/src/file_meta_inline.rs
+186
View File
@@ -0,0 +1,186 @@
use serde::{Deserialize, Serialize};
use crate::error::{Error, Result};
use std::io::{Cursor, Read};
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct InlineData(Vec<u8>);
const INLINE_DATA_VER: u8 = 1;
impl InlineData {
pub fn new() -> Self {
Self(Vec::new())
}
pub fn update(&mut self, buf: &[u8]) {
self.0 = buf.to_vec()
}
pub fn as_slice(&self) -> &[u8] {
&self.0.as_slice()
}
pub fn version_ok(&self) -> bool {
if self.0.is_empty() {
return true;
}
self.0[0] > 0 && self.0[0] <= INLINE_DATA_VER
}
pub fn after_version(&self) -> &[u8] {
if self.0.is_empty() {
&self.0
} else {
&self.0[1..]
}
}
pub fn find(&self, key: &str) -> Result<Option<Vec<u8>>> {
if self.0.is_empty() || !self.version_ok() {
return Ok(None);
}
let buf = self.after_version();
let mut cur = Cursor::new(buf);
let mut fields_len = rmp::decode::read_map_len(&mut cur)?;
while fields_len > 0 {
fields_len -= 1;
let str_len = rmp::decode::read_str_len(&mut cur)?;
let mut field_buff = vec![0u8; str_len as usize];
cur.read_exact(&mut field_buff)?;
let field = String::from_utf8(field_buff)?;
let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize;
let start = cur.position() as usize;
let end = start + bin_len;
cur.set_position(end as u64);
if field.as_str() == key {
let buf = &buf[start..end];
return Ok(Some(buf.to_vec()));
}
}
Ok(None)
}
fn validate(&self) -> Result<()> {
if self.0.is_empty() {
return Ok(());
}
let mut cur = Cursor::new(self.after_version());
let mut fields_len = rmp::decode::read_map_len(&mut cur)?;
while fields_len > 0 {
fields_len -= 1;
let str_len = rmp::decode::read_str_len(&mut cur)?;
let mut field_buff = vec![0u8; str_len as usize];
cur.read_exact(&mut field_buff)?;
let field = String::from_utf8(field_buff)?;
if field.is_empty() {
return Err(Error::msg("InlineData key empty"));
}
let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize;
let start = cur.position() as usize;
let end = start + bin_len;
cur.set_position(end as u64);
}
Ok(())
}
pub fn replace(&mut self, key: &str, value: Vec<u8>) -> Result<()> {
if self.after_version().is_empty() {
let mut keys = Vec::with_capacity(1);
let mut values = Vec::with_capacity(1);
keys.push(key.to_owned());
values.push(value);
return self.serialize(keys, values);
}
let buf = self.after_version();
let mut cur = Cursor::new(buf);
let mut fields_len = rmp::decode::read_map_len(&mut cur)? as usize;
let mut keys = Vec::with_capacity(fields_len + 1);
let mut values = Vec::with_capacity(fields_len + 1);
let mut replaced = false;
while fields_len > 0 {
fields_len -= 1;
let str_len = rmp::decode::read_str_len(&mut cur)?;
let mut field_buff = vec![0u8; str_len as usize];
cur.read_exact(&mut field_buff)?;
let find_key = String::from_utf8(field_buff)?;
let bin_len = rmp::decode::read_bin_len(&mut cur)? as usize;
let start = cur.position() as usize;
let end = start + bin_len;
cur.set_position(end as u64);
let find_value = &buf[start..end];
if find_key.as_str() == key {
values.push(value.clone());
replaced = true
} else {
values.push(find_value.to_vec());
}
keys.push(find_key);
}
if !replaced {
keys.push(key.to_owned());
values.push(value);
}
self.serialize(keys, values)
}
fn serialize(&mut self, keys: Vec<String>, values: Vec<Vec<u8>>) -> Result<()> {
assert_eq!(keys.len(), values.len(), "InlineData serialize: keys/values not match");
if keys.is_empty() {
self.0 = Vec::new();
return Ok(());
}
let mut wr = Vec::new();
wr.push(INLINE_DATA_VER);
let map_len = keys.len();
rmp::encode::write_map_len(&mut wr, map_len as u32)?;
for i in 0..map_len {
rmp::encode::write_str(&mut wr, keys[i].as_str())?;
rmp::encode::write_bin(&mut wr, values[i].as_slice())?;
}
self.0 = wr;
Ok(())
}
}
ecstore/src/global.rs
+22
View File
@@ -1,6 +1,7 @@
use lazy_static::lazy_static;
use std::{collections::HashMap, sync::Arc};
use tokio::sync::RwLock;
use uuid::Uuid;
use crate::{
disk::DiskStore,
@@ -8,6 +9,11 @@ use crate::{
store::ECStore,
};
pub const DISK_ASSUME_UNKNOWN_SIZE: u64 = 1 << 30;
pub const DISK_MIN_INODES: u64 = 1000;
pub const DISK_FILL_FRACTION: f64 = 0.99;
pub const DISK_RESERVE_FRACTION: f64 = 0.15;
lazy_static! {
pub static ref GLOBAL_OBJECT_API: Arc<RwLock<Option<ECStore>>> = Arc::new(RwLock::new(None));
pub static ref GLOBAL_LOCAL_DISK: Arc<RwLock<Vec<Option<DiskStore>>>> = Arc::new(RwLock::new(Vec::new()));
@@ -18,6 +24,17 @@ lazy_static! {
pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
pub static ref GLOBAL_Endpoints: RwLock<EndpointServerPools> = RwLock::new(EndpointServerPools(Vec::new()));
pub static ref GLOBAL_RootDiskThreshold: RwLock<u64> = RwLock::new(0);
static ref globalDeploymentIDPtr: RwLock<Uuid> = RwLock::new(Uuid::nil());
}
pub async fn set_global_deployment_id(id: Uuid) {
let mut id_ptr = globalDeploymentIDPtr.write().await;
*id_ptr = id
}
pub async fn get_global_deployment_id() -> Uuid {
let id_ptr = globalDeploymentIDPtr.read().await;
id_ptr.clone()
}
pub async fn set_global_endpoints(eps: Vec<PoolEndpoints>) {
@@ -39,6 +56,11 @@ pub async fn is_dist_erasure() -> bool {
*lock
}
pub async fn is_erasure_sd() -> bool {
let lock = GLOBAL_IsErasureSD.read().await;
*lock
}
pub async fn is_erasure() -> bool {
let lock = GLOBAL_IsErasure.read().await;
*lock
ecstore/src/heal/heal_ops.rs
+1 -1
View File
@@ -438,7 +438,7 @@ impl AllHealState {
_ = sleep(Duration::from_secs(5 * 60)) => {
let _ = self.mu.write().await;
let now = SystemTime::now();
let mut keys_to_reomve = Vec::new();
for (k, v) in self.heal_seq_map.iter() {
if v.has_ended().await && (UNIX_EPOCH + Duration::from_secs(*(v.end_time.read().await)) + KEEP_HEAL_SEQ_STATE_DURATION) < now {
ecstore/src/lib.rs
+5 -2
View File
@@ -1,7 +1,7 @@
pub mod bitrot;
pub mod cache_value;
mod chunk_stream;
mod config;
pub mod config;
pub mod disk;
pub mod disks_layout;
pub mod endpoints;
@@ -14,13 +14,16 @@ pub mod peer;
mod quorum;
pub mod set_disk;
mod sets;
mod storage_class;
pub mod store;
pub mod store_api;
mod store_init;
mod utils;
pub mod bucket;
pub mod file_meta_inline;
pub mod options;
pub(crate) mod store_err;
pub mod xhttp;
pub use global::is_legacy;
pub use global::new_object_layer_fn;
ecstore/src/options.rs
+71
View File
@@ -0,0 +1,71 @@
use http::{HeaderMap, HeaderValue};
use uuid::Uuid;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::error::{Error, Result};
use crate::store_api::ObjectOptions;
use crate::store_err::StorageError;
use crate::utils::path::is_dir_object;
use std::collections::HashMap;
pub async fn put_opts(
bucket: &str,
object: &str,
vid: Option<String>,
headers: &HeaderMap<HeaderValue>,
metadata: Option<HashMap<String, String>>,
) -> Result<ObjectOptions> {
let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await;
let version_suspended = BucketVersioningSys::prefix_suspended(bucket, object).await;
let vid = vid.map(|v| v.as_str().trim().to_owned());
if let Some(ref id) = vid {
if let Err(_err) = Uuid::parse_str(id.as_str()) {
return Err(Error::new(StorageError::InvalidVersionID(
bucket.to_owned(),
object.to_owned(),
id.clone(),
)));
}
if !versioned {
return Err(Error::new(StorageError::InvalidArgument(
bucket.to_owned(),
object.to_owned(),
id.clone(),
)));
}
}
let mut opts = put_opts_from_headers(headers, metadata)
.map_err(|err| Error::new(StorageError::InvalidArgument(bucket.to_owned(), object.to_owned(), err.to_string())))?;
opts.version_id = {
if is_dir_object(object) && vid.is_none() {
Some(Uuid::nil().to_string())
} else {
vid
}
};
opts.version_suspended = version_suspended;
opts.versioned = versioned;
Ok(opts)
}
pub fn put_opts_from_headers(
headers: &HeaderMap<HeaderValue>,
metadata: Option<HashMap<String, String>>,
) -> Result<ObjectOptions> {
// TODO custom headers
get_default_opts(headers, metadata, false)
}
fn get_default_opts(
_headers: &HeaderMap<HeaderValue>,
_metadata: Option<HashMap<String, String>>,
_copy_source: bool,
) -> Result<ObjectOptions> {
Ok(ObjectOptions::default())
}
ecstore/src/peer.rs
+2 -3
View File
@@ -258,9 +258,6 @@ impl PeerS3Client for LocalPeerS3Client {
}
}
warn!("list_bucket ress {:?}", &ress);
warn!("list_bucket errs {:?}", &errs);
let mut uniq_map: HashMap<&String, &VolumeInfo> = HashMap::new();
for info_list in ress.iter() {
@@ -279,6 +276,7 @@ impl PeerS3Client for LocalPeerS3Client {
.map(|&v| BucketInfo {
name: v.name.clone(),
created: v.created,
..Default::default()
})
.collect();
@@ -352,6 +350,7 @@ impl PeerS3Client for LocalPeerS3Client {
op.as_ref().map(|v| BucketInfo {
name: v.name.clone(),
created: v.created,
..Default::default()
})
})
.ok_or(Error::new(DiskError::VolumeNotFound))
ecstore/src/sets.rs
+67 -10
View File
@@ -12,11 +12,21 @@ use crate::{
disk::{
format::{DistributionAlgoVersion, FormatV3},
DiskStore,
}, endpoints::PoolEndpoints, error::{Error, Result}, global::{is_dist_erasure, GLOBAL_LOCAL_DISK_SET_DRIVES}, heal::{heal_commands::{HealOpts, HealResultItem}, heal_ops::HealObjectFn}, set_disk::SetDisks, store_api::{
},
endpoints::PoolEndpoints,
error::{Error, Result},
global::{is_dist_erasure, GLOBAL_LOCAL_DISK_SET_DRIVES},
heal::{
heal_commands::{HealOpts, HealResultItem},
heal_ops::HealObjectFn,
},
set_disk::SetDisks,
store_api::{
BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOptions, ObjectToDelete,
PartInfo, PutObjReader, StorageAPI,
}, utils::hash
ListMultipartsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectIO, ObjectInfo, ObjectOptions,
ObjectToDelete, PartInfo, PutObjReader, StorageAPI,
},
utils::hash,
};
use tokio::time::Duration;
@@ -153,6 +163,9 @@ impl Sets {
Ok(sets)
}
pub fn set_drive_count(&self) -> usize {
self.set_drive_count
}
pub async fn monitor_and_connect_endpoints(&self) {
tokio::time::sleep(Duration::from_secs(5)).await;
@@ -266,7 +279,7 @@ impl ObjectIO for Sets {
.get_object_reader(bucket, object, range, h, opts)
.await
}
async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
self.get_disks_by_key(object).put_object(bucket, object, data, opts).await
}
}
@@ -390,26 +403,61 @@ impl StorageAPI for Sets {
self.get_disks_by_key(object).get_object_info(bucket, object, opts).await
}
async fn put_object_info(&self, bucket: &str, object: &str, info: ObjectInfo, opts: &ObjectOptions) -> Result<()> {
async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<String> {
self.get_disks_by_key(object).get_object_tags(bucket, object, opts).await
}
#[tracing::instrument(level = "debug", skip(self))]
async fn put_object_tags(&self, bucket: &str, object: &str, tags: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
self.get_disks_by_key(object)
.put_object_info(bucket, object, info, opts)
.put_object_tags(bucket, object, tags, opts)
.await
}
async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
self.get_disks_by_key(object).delete_object_tags(bucket, object, opts).await
}
async fn copy_object_part(
&self,
_src_bucket: &str,
_src_object: &str,
_dst_bucket: &str,
_dst_object: &str,
_upload_id: &str,
_part_id: usize,
_start_offset: i64,
_length: i64,
_src_info: &ObjectInfo,
_src_opts: &ObjectOptions,
_dst_opts: &ObjectOptions,
) -> Result<()> {
unimplemented!()
}
async fn put_object_part(
&self,
bucket: &str,
object: &str,
upload_id: &str,
part_id: usize,
data: PutObjReader,
data: &mut PutObjReader,
opts: &ObjectOptions,
) -> Result<PartInfo> {
self.get_disks_by_key(object)
.put_object_part(bucket, object, upload_id, part_id, data, opts)
.await
}
async fn list_multipart_uploads(
&self,
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
max_uploads: usize,
) -> Result<ListMultipartsInfo> {
self.get_disks_by_key(&prefix)
.list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads)
.await
}
async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<MultipartUploadResult> {
self.get_disks_by_key(object).new_multipart_upload(bucket, object, opts).await
}
@@ -430,6 +478,13 @@ impl StorageAPI for Sets {
.complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts)
.await
}
async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result<Vec<Option<DiskStore>>> {
unimplemented!()
}
fn set_drive_counts(&self) -> Vec<usize> {
unimplemented!()
}
async fn delete_bucket(&self, _bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
unimplemented!()
@@ -441,7 +496,9 @@ impl StorageAPI for Sets {
unimplemented!()
}
async fn heal_object(&self, bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Result<HealResultItem> {
self.get_disks_by_key(object).heal_object(bucket, object, version_id, opts).await
self.get_disks_by_key(object)
.heal_object(bucket, object, version_id, opts)
.await
}
async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, func: HealObjectFn) -> Result<()> {
unimplemented!()
ecstore/src/storage_class.rs
-28
View File
@@ -1,28 +0,0 @@
// use crate::error::{Error, Result};
// default_partiy_count 默认配置，根据磁盘总数分配校验磁盘数量
pub fn default_partiy_count(drive: usize) -> usize {
match drive {
1 => 0,
2 | 3 => 1,
4 | 5 => 2,
6 | 7 => 3,
_ => 4,
}
}
// Define the minimum number of parity drives required.
// const MIN_PARITY_DRIVES: usize = 0;
// // ValidateParity validates standard storage class parity.
// pub fn validate_parity(ss_parity: usize, set_drive_count: usize) -> Result<()> {
// // if ss_parity > 0 && ss_parity < MIN_PARITY_DRIVES {
// // return Err(Error::msg(format!("parity {} 应该大于等于 {}", ss_parity, MIN_PARITY_DRIVES)));
// // }
// if ss_parity > set_drive_count / 2 {
// return Err(Error::msg(format!("parity {} 应该小于等于 {}", ss_parity, set_drive_count / 2)));
// }
// Ok(())
// }
ecstore/src/store.rs
+898 -43
View File
File diff suppressed because it is too large Load Diff
ecstore/src/store_api.rs
+298 -66
View File
@@ -1,10 +1,14 @@
use std::collections::HashMap;
use crate::{
disk::error::DiskError, error::{Error, Result}, heal::{
disk::DiskStore,
error::{Error, Result},
heal::{
heal_commands::{HealOpts, HealResultItem},
heal_ops::HealObjectFn,
}
},
utils::path::decode_dir_object,
xhttp,
};
use futures::StreamExt;
use http::HeaderMap;
@@ -22,44 +26,34 @@ pub const RESERVED_METADATA_PREFIX_LOWER: &str = "X-Rustfs-Internal-";
// #[derive(Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
pub struct FileInfo {
pub name: String,
pub volume: String,
pub name: String,
pub version_id: Option<Uuid>,
pub erasure: ErasureInfo,
pub is_latest: bool,
pub deleted: bool,
// DataDir of the file
// TransitionStatus
// TransitionedObjName
// TransitionTier
// TransitionVersionID
// ExpireRestored
pub data_dir: Option<Uuid>,
pub mod_time: Option<OffsetDateTime>,
pub size: usize,
pub data: Option<Vec<u8>>,
pub fresh: bool, // indicates this is a first time call to write FileInfo.
pub parts: Vec<ObjectPartInfo>,
pub is_latest: bool,
// #[serde(skip_serializing_if = "Option::is_none", default)]
pub tags: Option<HashMap<String, String>>,
// Mode
pub metadata: Option<HashMap<String, String>>,
pub parts: Vec<ObjectPartInfo>,
pub erasure: ErasureInfo,
// MarkDeleted
// ReplicationState
pub data: Option<Vec<u8>>,
pub num_versions: usize,
pub successor_mod_time: Option<OffsetDateTime>,
pub fresh: bool,
pub idx: usize,
// Checksum
pub versioned: bool,
}
// impl Default for FileInfo {
// fn default() -> Self {
// Self {
// version_id: Default::default(),
// erasure: Default::default(),
// deleted: Default::default(),
// data_dir: Default::default(),
// mod_time: None,
// size: Default::default(),
// data: Default::default(),
// fresh: Default::default(),
// name: Default::default(),
// volume: Default::default(),
// parts: Default::default(),
// is_latest: Default::default(),
// }
// }
// }
impl FileInfo {
pub fn new(object: &str, data_blocks: usize, parity_blocks: usize) -> Self {
let indexs = {
@@ -139,8 +133,16 @@ impl FileInfo {
Ok(t)
}
pub fn add_object_part(&mut self, num: usize, part_size: usize, mod_time: Option<OffsetDateTime>, actual_size: usize) {
pub fn add_object_part(
&mut self,
num: usize,
etag: Option<String>,
part_size: usize,
mod_time: Option<OffsetDateTime>,
actual_size: usize,
) {
let part = ObjectPartInfo {
etag,
number: num,
size: part_size,
mod_time,
@@ -159,23 +161,81 @@ impl FileInfo {
self.parts.sort_by(|a, b| a.number.cmp(&b.number));
}
pub fn to_object_info(&self, bucket: &str, object: &str, _versioned: bool) -> ObjectInfo {
pub fn to_object_info(&self, bucket: &str, object: &str, versioned: bool) -> ObjectInfo {
let name = decode_dir_object(object);
let mut version_id = self.version_id;
if versioned && version_id.is_none() {
version_id = Some(Uuid::nil())
}
let (content_type, content_encoding, etag) = {
if let Some(ref meta) = self.metadata {
let content_type = {
if let Some(ty) = meta.get("content-type") {
Some(ty.clone())
} else {
None
}
};
let content_encoding = {
if let Some(encoding) = meta.get("content-encoding") {
Some(encoding.clone())
} else {
None
}
};
let etag = {
if let Some(etag) = meta.get("etag") {
Some(etag.clone())
} else {
None
}
};
(content_type, content_encoding, etag)
} else {
(None, None, None)
}
};
let user_tags = self
.metadata
.as_ref()
.map(|m| {
if let Some(tags) = m.get(xhttp::AMZ_OBJECT_TAGGING) {
tags.clone()
} else {
"".to_string()
}
})
.unwrap_or_default();
let inlined = self.inline_data();
ObjectInfo {
bucket: bucket.to_string(),
name: object.to_string(),
name,
is_dir: object.starts_with('/'),
parity_blocks: self.erasure.parity_blocks,
data_blocks: self.erasure.data_blocks,
version_id: self.version_id,
version_id,
delete_marker: self.deleted,
mod_time: self.mod_time,
size: self.size,
parts: self.parts.clone(),
is_latest: self.is_latest,
tags: self.tags.clone(),
user_tags,
content_type,
content_encoding,
num_versions: self.num_versions,
successor_mod_time: self.successor_mod_time,
etag,
inlined,
..Default::default()
}
}
// to_part_offset 取offset 所在的part index, 返回part index, offset
pub fn to_part_offset(&self, offset: i64) -> Result<(usize, i64)> {
if offset == 0 {
@@ -194,11 +254,32 @@ impl FileInfo {
Err(Error::msg("part not found"))
}
pub fn set_inline_data(&mut self) {
if let Some(meta) = self.metadata.as_mut() {
meta.insert("x-rustfs-inline-data".to_owned(), "true".to_owned());
} else {
let mut meta = HashMap::new();
meta.insert("x-rustfs-inline-data".to_owned(), "true".to_owned());
self.metadata = Some(meta);
}
}
pub fn inline_data(&self) -> bool {
if let Some(ref meta) = self.metadata {
if let Some(val) = meta.get("x-rustfs-inline-data") {
val.as_str() == "true"
} else {
false
}
} else {
false
}
}
}
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
pub struct ObjectPartInfo {
// pub etag: Option<String>,
pub etag: Option<String>,
pub number: usize,
pub size: usize,
pub actual_size: usize, // 源数据大小
@@ -250,7 +331,10 @@ impl ErasureInfo {
}
}
ChecksumInfo {algorithm: DEFAULT_BITROT_ALGO, ..Default::default()}
ChecksumInfo {
algorithm: DEFAULT_BITROT_ALGO,
..Default::default()
}
}
// 算出每个分片大小
@@ -310,8 +394,20 @@ pub struct MakeBucketOptions {
pub no_lock: bool,
}
#[derive(Debug, Default, Clone)]
pub enum SRBucketDeleteOp {
#[default]
NoOp,
MarkDelete,
Purge,
}
#[derive(Debug, Default, Clone)]
pub struct DeleteBucketOptions {
pub no_lock: bool,
pub no_recreate: bool,
pub force: bool, // Force deletion
pub srdelete_op: SRBucketDeleteOp,
}
#[derive(Debug)]
@@ -448,8 +544,17 @@ pub struct ObjectOptions {
pub part_number: usize,
pub delete_prefix: bool,
pub version_id: String,
pub version_id: Option<String>,
pub no_lock: bool,
pub versioned: bool,
pub version_suspended: bool,
pub skip_decommissioned: bool,
pub skip_rebalancing: bool,
pub data_movement: bool,
pub src_pool_idx: usize,
}
// impl Default for ObjectOptions {
@@ -469,10 +574,13 @@ pub struct BucketOptions {
pub no_metadata: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct BucketInfo {
pub name: String,
pub created: Option<OffsetDateTime>,
pub deleted: Option<OffsetDateTime>,
pub versionning: bool,
pub object_locking: bool,
}
#[derive(Debug)]
@@ -485,9 +593,10 @@ pub struct PartInfo {
pub part_num: usize,
pub last_mod: Option<OffsetDateTime>,
pub size: usize,
pub etag: Option<String>,
}
#[derive(Debug)]
#[derive(Debug, Clone)]
pub struct CompletePart {
pub part_num: usize,
}
@@ -514,14 +623,22 @@ pub struct ObjectInfo {
pub data_blocks: usize,
pub version_id: Option<Uuid>,
pub delete_marker: bool,
pub user_tags: String,
pub parts: Vec<ObjectPartInfo>,
pub is_latest: bool,
pub tags: Option<HashMap<String, String>>,
pub content_type: Option<String>,
pub content_encoding: Option<String>,
pub num_versions: usize,
pub successor_mod_time: Option<OffsetDateTime>,
pub put_object_reader: Option<PutObjReader>,
pub etag: Option<String>,
pub inlined: bool,
}
impl ObjectInfo {
pub fn is_compressed(&self) -> bool {
self.user_defined.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX))
self.user_defined
.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX))
}
pub fn get_actual_size(&self) -> Result<usize> {
@@ -597,6 +714,71 @@ pub struct ListObjectsV2Info {
pub prefixes: Vec<String>,
}
#[derive(Debug, Clone, Default)]
pub struct MultipartInfo {
// Name of the bucket.
pub bucket: String,
// Name of the object.
pub object: String,
// Upload ID identifying the multipart upload whose parts are being listed.
pub upload_id: String,
// Date and time at which the multipart upload was initiated.
pub initiated: Option<OffsetDateTime>,
// Any metadata set during InitMultipartUpload, including encryption headers.
pub user_defined: HashMap<String, String>,
}
// ListMultipartsInfo - represents bucket resources for incomplete multipart uploads.
#[derive(Debug, Clone, Default)]
pub struct ListMultipartsInfo {
// Together with upload-id-marker, this parameter specifies the multipart upload
// after which listing should begin.
pub key_marker: String,
// Together with key-marker, specifies the multipart upload after which listing
// should begin. If key-marker is not specified, the upload-id-marker parameter
// is ignored.
pub upload_id_marker: String,
// When a list is truncated, this element specifies the value that should be
// used for the key-marker request parameter in a subsequent request.
pub next_key_marker: String,
// When a list is truncated, this element specifies the value that should be
// used for the upload-id-marker request parameter in a subsequent request.
pub next_upload_id_marker: String,
// Maximum number of multipart uploads that could have been included in the
// response.
pub max_uploads: usize,
// Indicates whether the returned list of multipart uploads is truncated. A
// value of true indicates that the list was truncated. The list can be truncated
// if the number of multipart uploads exceeds the limit allowed or specified
// by max uploads.
pub is_truncated: bool,
// List of all pending uploads.
pub uploads: Vec<MultipartInfo>,
// When a prefix is provided in the request, The result contains only keys
// starting with the specified prefix.
pub prefix: String,
// A character used to truncate the object prefixes.
// NOTE: only supported delimiter is '/'.
pub delimiter: String,
// CommonPrefixes contains all (if there are any) keys between Prefix and the
// next occurrence of the string specified by delimiter.
pub common_prefixes: Vec<String>,
// encoding_type: String, // Not supported yet.
}
#[derive(Debug, Default, Clone)]
pub struct ObjectToDelete {
pub object_name: String,
@@ -616,6 +798,7 @@ pub struct DeletedObject {
#[async_trait::async_trait]
pub trait ObjectIO: Send + Sync + 'static {
// GetObjectNInfo
async fn get_object_reader(
&self,
bucket: &str,
@@ -624,23 +807,24 @@ pub trait ObjectIO: Send + Sync + 'static {
h: HeaderMap,
opts: &ObjectOptions,
) -> Result<GetObjectReader>;
async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo>;
// PutObject
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo>;
}
#[async_trait::async_trait]
pub trait StorageAPI: ObjectIO {
// NewNSLock
// Shutdown
// NSScanner
// BackendInfo
// StorageInfo
// LocalStorageInfo
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo>;
async fn delete_objects(
&self,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> Result<(Vec<DeletedObject>, Vec<Option<Error>>)>;
#[warn(clippy::too_many_arguments)]
async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
// ListObjects
async fn list_objects_v2(
&self,
bucket: &str,
@@ -651,29 +835,61 @@ pub trait StorageAPI: ObjectIO {
fetch_owner: bool,
start_after: &str,
) -> Result<ListObjectsV2Info>;
// ListObjectVersions
// Walk
// GetObjectNInfo ObjectIO
async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo>;
// PutObject ObjectIO
// CopyObject
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo>;
async fn delete_objects(
&self,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> Result<(Vec<DeletedObject>, Vec<Option<Error>>)>;
#[warn(clippy::too_many_arguments)]
// TransitionObject
// RestoreTransitionedObject
async fn put_object_info(&self, bucket: &str, object: &str, info: ObjectInfo, opts: &ObjectOptions) -> Result<()>;
// async fn get_object_reader(
// &self,
// bucket: &str,
// object: &str,
// range: HTTPRangeSpec,
// h: HeaderMap,
// opts: &ObjectOptions,
// ) -> Result<GetObjectReader>;
// async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo>;
// ListMultipartUploads
async fn list_multipart_uploads(
&self,
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
max_uploads: usize,
) -> Result<ListMultipartsInfo>;
async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<MultipartUploadResult>;
// CopyObjectPart
async fn copy_object_part(
&self,
src_bucket: &str,
src_object: &str,
dst_bucket: &str,
dst_object: &str,
upload_id: &str,
part_id: usize,
start_offset: i64,
length: i64,
src_info: &ObjectInfo,
src_opts: &ObjectOptions,
dst_opts: &ObjectOptions,
) -> Result<()>;
async fn put_object_part(
&self,
bucket: &str,
object: &str,
upload_id: &str,
part_id: usize,
data: PutObjReader,
data: &mut PutObjReader,
opts: &ObjectOptions,
) -> Result<PartInfo>;
async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<MultipartUploadResult>;
// GetMultipartInfo
// ListObjectParts
async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()>;
async fn complete_multipart_upload(
&self,
@@ -683,6 +899,22 @@ pub trait StorageAPI: ObjectIO {
uploaded_parts: Vec<CompletePart>,
opts: &ObjectOptions,
) -> Result<ObjectInfo>;
// GetDisks
async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result<Vec<Option<DiskStore>>>;
// SetDriveCounts
fn set_drive_counts(&self) -> Vec<usize>;
// HealFormat
// HealBucket
// HealObject
// HealObjects
// CheckAbandonedParts
// Health
// PutObjectMetadata
// DecomTieredObject
async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<String>;
async fn put_object_tags(&self, bucket: &str, object: &str, tags: &str, opts: &ObjectOptions) -> Result<ObjectInfo>;
async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo>;
async fn heal_format(&self, dry_run: bool) -> Result<HealResultItem>;
async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem>;
async fn heal_object(&self, bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Result<HealResultItem>;
ecstore/src/store_err.rs
+97
View File
@@ -0,0 +1,97 @@
use crate::{disk::error::is_err_file_not_found, error::Error};
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum StorageError {
#[error("Invalid arguments provided for {0}/{1}-{2}")]
InvalidArgument(String, String, String),
#[error("Bucket name invalid: {0}")]
BucketNameInvalid(String),
#[error("Object name invalid: {0}/{1}")]
ObjectNameInvalid(String, String),
#[error("Bucket exists: {0}")]
BucketExists(String),
#[error("Invalid UploadID KeyCombination: {0}/{1}")]
InvalidUploadIDKeyCombination(String, String),
#[error("Malformed UploadID: {0}")]
MalformedUploadID(String),
#[error("Object name too long: {0}/{1}")]
ObjectNameTooLong(String, String),
#[error("Object name contains forward slash as prefix: {0}/{1}")]
ObjectNamePrefixAsSlash(String, String),
#[error("Object not found: {0}/{1}")]
ObjectNotFound(String, String),
#[error("Version not found: {0}/{1}-{2}")]
VersionNotFound(String, String, String),
#[error("Invalid upload id: {0}/{1}-{2}")]
InvalidUploadID(String, String, String),
#[error("Invalid version id: {0}/{1}-{2}")]
InvalidVersionID(String, String, String),
#[error("invalid data movement operation, source and destination pool are the same for : {0}/{1}-{2}")]
DataMovementOverwriteErr(String, String, String),
}
pub fn is_err_invalid_upload_id(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() {
match e {
StorageError::InvalidUploadID(_, _, _) => true,
_ => false,
}
} else {
false
}
}
pub fn is_err_version_not_found(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() {
match e {
StorageError::VersionNotFound(_, _, _) => true,
_ => false,
}
} else {
false
}
}
pub fn is_err_bucket_exists(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() {
match e {
StorageError::BucketExists(_) => true,
_ => false,
}
} else {
false
}
}
pub fn is_err_object_not_found(err: &Error) -> bool {
if is_err_file_not_found(err) {
return true;
}
if let Some(e) = err.downcast_ref::<StorageError>() {
match e {
StorageError::ObjectNotFound(_, _) => true,
_ => false,
}
} else {
false
}
}
#[test]
fn test_storage_error() {
let e1 = Error::new(StorageError::BucketExists("ss".into()));
let e2 = Error::new(StorageError::ObjectNotFound("ss".into(), "sdf".to_owned()));
assert_eq!(is_err_bucket_exists(&e1), true);
assert_eq!(is_err_object_not_found(&e1), false);
assert_eq!(is_err_object_not_found(&e2), true);
}
ecstore/src/store_init.rs
+6
View File
@@ -1,3 +1,4 @@
use crate::config::{storageclass, KVS};
use crate::{
disk::{
error::DiskError,
@@ -283,6 +284,11 @@ async fn save_format_file(disk: &Option<DiskStore>, format: &Option<FormatV3>) -
Ok(())
}
pub fn ec_drives_no_config(set_drive_count: usize) -> Result<usize> {
let sc = storageclass::lookup_config(&KVS::new(), set_drive_count)?;
Ok(sc.get_parity_for_sc(storageclass::STANDARD).unwrap_or_default())
}
#[derive(Debug, thiserror::Error)]
pub enum ErasureError {
#[error("erasure read quorum")]
ecstore/src/utils/hasher.rs
+45
View File
@@ -0,0 +1,45 @@
use sha2::{
digest::{Reset, Update},
Digest, Sha256 as sha_sha256,
};
trait Hasher {
fn write(&mut self, bytes: &[u8]);
fn reset(&mut self);
fn sum(&mut self) -> impl AsRef<[u8]>;
fn size(&self) -> usize;
fn block_size(&self) -> usize;
}
struct Sha256 {
hasher: sha_sha256,
}
impl Sha256 {
pub fn new() -> Self {
Self {
hasher: sha_sha256::new(),
}
}
}
impl Hasher for Sha256 {
fn write(&mut self, bytes: &[u8]) {
Update::update(&mut self.hasher, bytes);
}
fn reset(&mut self) {
Reset::reset(&mut self.hasher);
}
fn sum(&mut self) -> impl AsRef<[u8]> {
self.hasher.clone().finalize()
}
fn size(&self) -> usize {
32
}
fn block_size(&self) -> usize {
64
}
}
ecstore/src/utils/mod.rs
+2 -1
View File
@@ -2,7 +2,8 @@ pub mod crypto;
pub mod ellipses;
pub mod fs;
pub mod hash;
pub mod hasher;
pub mod net;
pub mod path;
pub mod os;
pub mod path;
pub mod wildcard;
ecstore/src/utils/os/linux.rs
+1 -1
View File
@@ -109,4 +109,4 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result<bool> {
let stat2 = stat(disk2)?;
Ok(stat1.st_dev == stat2.st_dev)
}
}
ecstore/src/utils/os/mod.rs
+5 -4
View File
@@ -6,9 +6,10 @@ mod unix;
mod windows;
#[cfg(target_os = "linux")]
pub use linux::get_info;
pub use linux::same_disk;
pub use linux::{get_info, same_disk};
// pub use linux::same_disk;
#[cfg(all(unix, not(target_os = "linux")))]
pub use unix::get_info;
pub use unix::{get_info, same_disk};
#[cfg(target_os = "windows")]
pub use windows::get_info;
pub use windows::{get_info, same_disk};
ecstore/src/utils/os/unix.rs
+3 -3
View File
@@ -1,5 +1,5 @@
use crate::disk::Info;
use nix::sys::{statfs::statfs, stat::stat};
use crate::{disk::Info, error::Result};
use nix::sys::{stat::stat, statfs::statfs};
use std::io::{Error, ErrorKind};
use std::path::Path;
@@ -74,4 +74,4 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result<bool> {
let stat2 = stat(disk2)?;
Ok(stat1.st_dev == stat2.st_dev)
}
}
ecstore/src/utils/os/windows.rs
+2 -2
View File
@@ -1,4 +1,4 @@
use crate::disk::Info;
use crate::{disk::Info, error::Result};
use std::io::{Error, ErrorKind, Result};
use std::mem;
use std::os::windows::ffi::OsStrExt;
@@ -133,4 +133,4 @@ fn get_fs_type(p: &[WCHAR]) -> Result<String> {
pub fn same_disk(disk1: &str, disk2: &str) -> Result<bool> {
Ok(false)
}
}
ecstore/src/utils/path.rs
+8 -1
View File
@@ -4,6 +4,8 @@ const GLOBAL_DIR_SUFFIX: &str = "__XLDIR__";
pub const SLASH_SEPARATOR: &str = "/";
pub const GLOBAL_DIR_SUFFIX_WITH_SLASH: &str = "__XLDIR__/";
pub fn has_suffix(s: &str, suffix: &str) -> bool {
if cfg!(target_os = "windows") {
s.to_lowercase().ends_with(&suffix.to_lowercase())
@@ -20,6 +22,11 @@ pub fn encode_dir_object(object: &str) -> String {
}
}
pub fn is_dir_object(object: &str) -> bool {
let obj = encode_dir_object(object);
obj.ends_with(GLOBAL_DIR_SUFFIX)
}
#[allow(dead_code)]
pub fn decode_dir_object(object: &str) -> String {
if has_suffix(object, GLOBAL_DIR_SUFFIX) {
@@ -54,7 +61,7 @@ pub fn has_profix(s: &str, prefix: &str) -> bool {
pub fn path_join(elem: &[PathBuf]) -> PathBuf {
let mut joined_path = PathBuf::new();
for path in elem {
joined_path.push(path);
}
ecstore/src/utils/stat_linux.rs
+84
View File
@@ -0,0 +1,84 @@
use nix::sys::{
stat::{major, minor, stat},
statfs::{statfs, FsType},
};
use crate::{
disk::Info,
error::{Error, Result},
};
use lazy_static::lazy_static;
use std::collections::HashMap;
lazy_static! {
static ref FS_TYPE_TO_STRING_MAP: HashMap<&'static str, &'static str> = {
let mut m = HashMap::new();
m.insert("1021994", "TMPFS");
m.insert("137d", "EXT");
m.insert("4244", "HFS");
m.insert("4d44", "MSDOS");
m.insert("52654973", "REISERFS");
m.insert("5346544e", "NTFS");
m.insert("58465342", "XFS");
m.insert("61756673", "AUFS");
m.insert("6969", "NFS");
m.insert("ef51", "EXT2OLD");
m.insert("ef53", "EXT4");
m.insert("f15f", "ecryptfs");
m.insert("794c7630", "overlayfs");
m.insert("2fc12fc1", "zfs");
m.insert("ff534d42", "cifs");
m.insert("53464846", "wslfs");
m
};
}
fn get_fs_type(ftype: FsType) -> String {
let binding = format!("{:?}", ftype);
let fs_type_hex = binding.as_str();
match FS_TYPE_TO_STRING_MAP.get(fs_type_hex) {
Some(fs_type_string) => fs_type_string.to_string(),
None => "UNKNOWN".to_string(),
}
}
pub fn get_info(path: &str, first_time: bool) -> Result<Info> {
let statfs = statfs(path)?;
let reserved_blocks = statfs.blocks_free() - statfs.blocks_available();
let mut info = Info {
total: statfs.block_size() as u64 * (statfs.blocks() - reserved_blocks),
free: statfs.blocks() as u64 * statfs.blocks_available(),
files: statfs.files(),
ffree: statfs.files_free(),
fstype: get_fs_type(statfs.filesystem_type()),
..Default::default()
};
let stat = stat(path)?;
let dev_id = stat.st_dev as u64;
info.major = major(dev_id);
info.minor = minor(dev_id);
if info.free > info.total {
return Err(Error::from_string(format!(
"detected free space {} > total drive space {}, fs corruption at {}. please run 'fsck'",
info.free, info.total, path
)));
}
info.used = info.total - info.free;
if first_time {
// todo
}
Ok(info)
}
pub fn same_disk(disk1: &str, disk2: &str) -> Result<bool> {
let stat1 = stat(disk1)?;
let stat2 = stat(disk2)?;
Ok(stat1.st_dev == stat2.st_dev)
}
ecstore/src/xhttp.rs
+1
View File
@@ -0,0 +1 @@
pub(crate) const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";