gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-27 15:37:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #116 from rustfs/reader

Browse Source 
Reader
This commit is contained in:
weisd
2024-11-05 09:43:20 +08:00
committed by GitHub
parent 692048d537 e922c70c99
commit 0ae379fd7a
26 changed files with 655 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -1851,6 +1851,7 @@ dependencies = [
"time", "time",
"tokio", "tokio",
"tokio-stream", "tokio-stream",
"tokio-util",
"tonic", "tonic",
"tonic-build", "tonic-build",
"tonic-reflection", "tonic-reflection",
@@ -2333,6 +2334,7 @@ checksum = "61e7c3654c13bcd040d4a03abee2c75b1d14a37b423cf5a813ceae1cc903ec6a"
dependencies = [ dependencies = [
"bytes", "bytes",
"futures-core", "futures-core",
"futures-io",
"futures-sink", "futures-sink",
"pin-project-lite", "pin-project-lite",
"tokio", "tokio",
ecstore/Cargo.toml
+1 -1
View File
@@ -39,7 +39,7 @@ nix = { version = "0.29.0", features = ["fs"] }
path-absolutize = "3.1.1" path-absolutize = "3.1.1"
protos.workspace = true protos.workspace = true
rmp-serde = "1.3.0" rmp-serde = "1.3.0"
tokio-util = { version = "0.7.12", features = ["io"] } tokio-util = { version = "0.7.12", features = ["io", "compat"] }
crc32fast = "1.4.2" crc32fast = "1.4.2"
siphasher = "1.0.1" siphasher = "1.0.1"
base64-simd = "0.8.0" base64-simd = "0.8.0"
ecstore/src/bitrot.rs
+2 -2
View File
@@ -1,5 +1,5 @@
use crate::{ use crate::{
disk::{error::DiskError, DiskStore, FileReader, FileWriter, Reader}, disk::{error::DiskError, DiskAPI, DiskStore, FileReader, FileWriter, Reader},
erasure::{ReadAt, Writer}, erasure::{ReadAt, Writer},
error::{Error, Result}, error::{Error, Result},
store_api::BitrotAlgorithm, store_api::BitrotAlgorithm,
@@ -612,7 +612,7 @@ mod test {
use crate::{ use crate::{
bitrot::{new_bitrot_writer, BITROT_ALGORITHMS}, bitrot::{new_bitrot_writer, BITROT_ALGORITHMS},
disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption}, disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskAPI, DiskOption},
error::{Error, Result}, error::{Error, Result},
store_api::BitrotAlgorithm, store_api::BitrotAlgorithm,
}; };
ecstore/src/bucket/metadata_sys.rs
+1 -1
View File
@@ -60,7 +60,7 @@ pub async fn delete(bucket: &str, config_file: &str) -> Result<OffsetDateTime> {
let bucket_meta_sys_lock = get_bucket_metadata_sys().await; let bucket_meta_sys_lock = get_bucket_metadata_sys().await;
let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; let mut bucket_meta_sys = bucket_meta_sys_lock.write().await;
bucket_meta_sys.delete(&bucket, config_file).await bucket_meta_sys.delete(bucket, config_file).await
} }
pub async fn get_tagging_config(bucket: &str) -> Result<(Tagging, OffsetDateTime)> { pub async fn get_tagging_config(bucket: &str) -> Result<(Tagging, OffsetDateTime)> {
ecstore/src/bucket/policy/action.rs
+133 -62
View File
@@ -3,6 +3,7 @@ use lazy_static::lazy_static;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::{ use std::{
collections::{HashMap, HashSet}, collections::{HashMap, HashSet},
str::FromStr,
vec, vec,
}; };
@@ -282,69 +283,139 @@ impl Action {
} }
} }
pub 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),
// "s3:DeleteBucket" => Some(Action::DeleteBucket),
// "s3:ForceDeleteBucket" => Some(Action::ForceDeleteBucket),
// "s3:DeleteBucketPolicy" => Some(Action::DeleteBucketPolicy),
// "s3:DeleteBucketCors" => Some(Action::DeleteBucketCors),
// "s3:DeleteObject" => Some(Action::DeleteObject),
// "s3:GetBucketLocation" => Some(Action::GetBucketLocation),
// "s3:GetBucketNotification" => Some(Action::GetBucketNotification),
// "s3:GetBucketPolicy" => Some(Action::GetBucketPolicy),
// "s3:GetBucketCors" => Some(Action::GetBucketCors),
// "s3:GetObject" => Some(Action::GetObject),
// "s3:GetObjectAttributes" => Some(Action::GetObjectAttributes),
// "s3:HeadBucket" => Some(Action::HeadBucket),
// "s3:ListAllMyBuckets" => Some(Action::ListAllMyBuckets),
// "s3:ListBucket" => Some(Action::ListBucket),
// "s3:GetBucketPolicyStatus" => Some(Action::GetBucketPolicyStatus),
// "s3:ListBucketVersions" => Some(Action::ListBucketVersions),
// "s3:ListBucketMultipartUploads" => Some(Action::ListBucketMultipartUploads),
// "s3:ListenNotification" => Some(Action::ListenNotification),
// "s3:ListenBucketNotification" => Some(Action::ListenBucketNotification),
// "s3:ListMultipartUploadParts" => Some(Action::ListMultipartUploadParts),
// "s3:PutLifecycleConfiguration" => Some(Action::PutLifecycleConfiguration),
// "s3:GetLifecycleConfiguration" => Some(Action::GetLifecycleConfiguration),
// "s3:PutBucketNotification" => Some(Action::PutBucketNotification),
// "s3:PutBucketPolicy" => Some(Action::PutBucketPolicy),
// "s3:PutBucketCors" => Some(Action::PutBucketCors),
// "s3:PutObject" => Some(Action::PutObject),
// "s3:DeleteObjectVersion" => Some(Action::DeleteObjectVersion),
// "s3:DeleteObjectVersionTagging" => Some(Action::DeleteObjectVersionTagging),
// "s3:GetObjectVersion" => Some(Action::GetObjectVersion),
// "s3:GetObjectVersionAttributes" => Some(Action::GetObjectVersionAttributes),
// "s3:GetObjectVersionTagging" => Some(Action::GetObjectVersionTagging),
// "s3:PutObjectVersionTagging" => Some(Action::PutObjectVersionTagging),
// "s3:BypassGovernanceRetention" => Some(Action::BypassGovernanceRetention),
// "s3:PutObjectRetention" => Some(Action::PutObjectRetention),
// "s3:GetObjectRetention" => Some(Action::GetObjectRetention),
// "s3:GetObjectLegalHold" => Some(Action::GetObjectLegalHold),
// "s3:PutObjectLegalHold" => Some(Action::PutObjectLegalHold),
// "s3:GetBucketObjectLockConfiguration" => Some(Action::GetBucketObjectLockConfiguration),
// "s3:PutBucketObjectLockConfiguration" => Some(Action::PutBucketObjectLockConfiguration),
// "s3:GetBucketTagging" => Some(Action::GetBucketTagging),
// "s3:PutBucketTagging" => Some(Action::PutBucketTagging),
// "s3:GetObjectTagging" => Some(Action::GetObjectTagging),
// "s3:PutObjectTagging" => Some(Action::PutObjectTagging),
// "s3:DeleteObjectTagging" => Some(Action::DeleteObjectTagging),
// "s3:PutEncryptionConfiguration" => Some(Action::PutBucketEncryption),
// "s3:GetEncryptionConfiguration" => Some(Action::GetBucketEncryption),
// "s3:PutBucketVersioning" => Some(Action::PutBucketVersioning),
// "s3:GetBucketVersioning" => Some(Action::GetBucketVersioning),
// "s3:PutReplicationConfiguration" => Some(Action::PutReplicationConfiguration),
// "s3:GetReplicationConfiguration" => Some(Action::GetReplicationConfiguration),
// "s3:ReplicateObject" => Some(Action::ReplicateObject),
// "s3:ReplicateDelete" => Some(Action::ReplicateDelete),
// "s3:ReplicateTags" => Some(Action::ReplicateTags),
// "s3:GetObjectVersionForReplication" => Some(Action::GetObjectVersionForReplication),
// "s3:RestoreObject" => Some(Action::RestoreObject),
// "s3:ResetBucketReplicationState" => Some(Action::ResetBucketReplicationState),
// "s3:PutObjectFanOut" => Some(Action::PutObjectFanOut),
// "s3:*" => Some(Action::AllActions),
// _ => None,
// }
// }
}
impl FromStr for Action {
type Err = ();
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s { match s {
"s3:AbortMultipartUpload" => Some(Action::AbortMultipartUpload), "s3:AbortMultipartUpload" => Ok(Action::AbortMultipartUpload),
"s3:CreateBucket" => Some(Action::CreateBucket), "s3:CreateBucket" => Ok(Action::CreateBucket),
"s3:DeleteBucket" => Some(Action::DeleteBucket), "s3:DeleteBucket" => Ok(Action::DeleteBucket),
"s3:ForceDeleteBucket" => Some(Action::ForceDeleteBucket), "s3:ForceDeleteBucket" => Ok(Action::ForceDeleteBucket),
"s3:DeleteBucketPolicy" => Some(Action::DeleteBucketPolicy), "s3:DeleteBucketPolicy" => Ok(Action::DeleteBucketPolicy),
"s3:DeleteBucketCors" => Some(Action::DeleteBucketCors), "s3:DeleteBucketCors" => Ok(Action::DeleteBucketCors),
"s3:DeleteObject" => Some(Action::DeleteObject), "s3:DeleteObject" => Ok(Action::DeleteObject),
"s3:GetBucketLocation" => Some(Action::GetBucketLocation), "s3:GetBucketLocation" => Ok(Action::GetBucketLocation),
"s3:GetBucketNotification" => Some(Action::GetBucketNotification), "s3:GetBucketNotification" => Ok(Action::GetBucketNotification),
"s3:GetBucketPolicy" => Some(Action::GetBucketPolicy), "s3:GetBucketPolicy" => Ok(Action::GetBucketPolicy),
"s3:GetBucketCors" => Some(Action::GetBucketCors), "s3:GetBucketCors" => Ok(Action::GetBucketCors),
"s3:GetObject" => Some(Action::GetObject), "s3:GetObject" => Ok(Action::GetObject),
"s3:GetObjectAttributes" => Some(Action::GetObjectAttributes), "s3:GetObjectAttributes" => Ok(Action::GetObjectAttributes),
"s3:HeadBucket" => Some(Action::HeadBucket), "s3:HeadBucket" => Ok(Action::HeadBucket),
"s3:ListAllMyBuckets" => Some(Action::ListAllMyBuckets), "s3:ListAllMyBuckets" => Ok(Action::ListAllMyBuckets),
"s3:ListBucket" => Some(Action::ListBucket), "s3:ListBucket" => Ok(Action::ListBucket),
"s3:GetBucketPolicyStatus" => Some(Action::GetBucketPolicyStatus), "s3:GetBucketPolicyStatus" => Ok(Action::GetBucketPolicyStatus),
"s3:ListBucketVersions" => Some(Action::ListBucketVersions), "s3:ListBucketVersions" => Ok(Action::ListBucketVersions),
"s3:ListBucketMultipartUploads" => Some(Action::ListBucketMultipartUploads), "s3:ListBucketMultipartUploads" => Ok(Action::ListBucketMultipartUploads),
"s3:ListenNotification" => Some(Action::ListenNotification), "s3:ListenNotification" => Ok(Action::ListenNotification),
"s3:ListenBucketNotification" => Some(Action::ListenBucketNotification), "s3:ListenBucketNotification" => Ok(Action::ListenBucketNotification),
"s3:ListMultipartUploadParts" => Some(Action::ListMultipartUploadParts), "s3:ListMultipartUploadParts" => Ok(Action::ListMultipartUploadParts),
"s3:PutLifecycleConfiguration" => Some(Action::PutLifecycleConfiguration), "s3:PutLifecycleConfiguration" => Ok(Action::PutLifecycleConfiguration),
"s3:GetLifecycleConfiguration" => Some(Action::GetLifecycleConfiguration), "s3:GetLifecycleConfiguration" => Ok(Action::GetLifecycleConfiguration),
"s3:PutBucketNotification" => Some(Action::PutBucketNotification), "s3:PutBucketNotification" => Ok(Action::PutBucketNotification),
"s3:PutBucketPolicy" => Some(Action::PutBucketPolicy), "s3:PutBucketPolicy" => Ok(Action::PutBucketPolicy),
"s3:PutBucketCors" => Some(Action::PutBucketCors), "s3:PutBucketCors" => Ok(Action::PutBucketCors),
"s3:PutObject" => Some(Action::PutObject), "s3:PutObject" => Ok(Action::PutObject),
"s3:DeleteObjectVersion" => Some(Action::DeleteObjectVersion), "s3:DeleteObjectVersion" => Ok(Action::DeleteObjectVersion),
"s3:DeleteObjectVersionTagging" => Some(Action::DeleteObjectVersionTagging), "s3:DeleteObjectVersionTagging" => Ok(Action::DeleteObjectVersionTagging),
"s3:GetObjectVersion" => Some(Action::GetObjectVersion), "s3:GetObjectVersion" => Ok(Action::GetObjectVersion),
"s3:GetObjectVersionAttributes" => Some(Action::GetObjectVersionAttributes), "s3:GetObjectVersionAttributes" => Ok(Action::GetObjectVersionAttributes),
"s3:GetObjectVersionTagging" => Some(Action::GetObjectVersionTagging), "s3:GetObjectVersionTagging" => Ok(Action::GetObjectVersionTagging),
"s3:PutObjectVersionTagging" => Some(Action::PutObjectVersionTagging), "s3:PutObjectVersionTagging" => Ok(Action::PutObjectVersionTagging),
"s3:BypassGovernanceRetention" => Some(Action::BypassGovernanceRetention), "s3:BypassGovernanceRetention" => Ok(Action::BypassGovernanceRetention),
"s3:PutObjectRetention" => Some(Action::PutObjectRetention), "s3:PutObjectRetention" => Ok(Action::PutObjectRetention),
"s3:GetObjectRetention" => Some(Action::GetObjectRetention), "s3:GetObjectRetention" => Ok(Action::GetObjectRetention),
"s3:GetObjectLegalHold" => Some(Action::GetObjectLegalHold), "s3:GetObjectLegalHold" => Ok(Action::GetObjectLegalHold),
"s3:PutObjectLegalHold" => Some(Action::PutObjectLegalHold), "s3:PutObjectLegalHold" => Ok(Action::PutObjectLegalHold),
"s3:GetBucketObjectLockConfiguration" => Some(Action::GetBucketObjectLockConfiguration), "s3:GetBucketObjectLockConfiguration" => Ok(Action::GetBucketObjectLockConfiguration),
"s3:PutBucketObjectLockConfiguration" => Some(Action::PutBucketObjectLockConfiguration), "s3:PutBucketObjectLockConfiguration" => Ok(Action::PutBucketObjectLockConfiguration),
"s3:GetBucketTagging" => Some(Action::GetBucketTagging), "s3:GetBucketTagging" => Ok(Action::GetBucketTagging),
"s3:PutBucketTagging" => Some(Action::PutBucketTagging), "s3:PutBucketTagging" => Ok(Action::PutBucketTagging),
"s3:GetObjectTagging" => Some(Action::GetObjectTagging), "s3:GetObjectTagging" => Ok(Action::GetObjectTagging),
"s3:PutObjectTagging" => Some(Action::PutObjectTagging), "s3:PutObjectTagging" => Ok(Action::PutObjectTagging),
"s3:DeleteObjectTagging" => Some(Action::DeleteObjectTagging), "s3:DeleteObjectTagging" => Ok(Action::DeleteObjectTagging),
"s3:PutEncryptionConfiguration" => Some(Action::PutBucketEncryption), "s3:PutEncryptionConfiguration" => Ok(Action::PutBucketEncryption),
"s3:GetEncryptionConfiguration" => Some(Action::GetBucketEncryption), "s3:GetEncryptionConfiguration" => Ok(Action::GetBucketEncryption),
"s3:PutBucketVersioning" => Some(Action::PutBucketVersioning), "s3:PutBucketVersioning" => Ok(Action::PutBucketVersioning),
"s3:GetBucketVersioning" => Some(Action::GetBucketVersioning), "s3:GetBucketVersioning" => Ok(Action::GetBucketVersioning),
"s3:PutReplicationConfiguration" => Some(Action::PutReplicationConfiguration), "s3:PutReplicationConfiguration" => Ok(Action::PutReplicationConfiguration),
"s3:GetReplicationConfiguration" => Some(Action::GetReplicationConfiguration), "s3:GetReplicationConfiguration" => Ok(Action::GetReplicationConfiguration),
"s3:ReplicateObject" => Some(Action::ReplicateObject), "s3:ReplicateObject" => Ok(Action::ReplicateObject),
"s3:ReplicateDelete" => Some(Action::ReplicateDelete), "s3:ReplicateDelete" => Ok(Action::ReplicateDelete),
"s3:ReplicateTags" => Some(Action::ReplicateTags), "s3:ReplicateTags" => Ok(Action::ReplicateTags),
"s3:GetObjectVersionForReplication" => Some(Action::GetObjectVersionForReplication), "s3:GetObjectVersionForReplication" => Ok(Action::GetObjectVersionForReplication),
"s3:RestoreObject" => Some(Action::RestoreObject), "s3:RestoreObject" => Ok(Action::RestoreObject),
"s3:ResetBucketReplicationState" => Some(Action::ResetBucketReplicationState), "s3:ResetBucketReplicationState" => Ok(Action::ResetBucketReplicationState),
"s3:PutObjectFanOut" => Some(Action::PutObjectFanOut), "s3:PutObjectFanOut" => Ok(Action::PutObjectFanOut),
"s3:*" => Some(Action::AllActions), "s3:*" => Ok(Action::AllActions),
_ => None, _ => Err(()),
} }
} }
} }
ecstore/src/cache_value/metacache_set.rs
+1 -1
View File
@@ -10,7 +10,7 @@ use tokio::{
}; };
use crate::{ use crate::{
disk::{DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions}, disk::{DiskAPI, DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions},
error::{Error, Result}, error::{Error, Result},
}; };
ecstore/src/config/common.rs
+3 -5
View File
@@ -173,11 +173,9 @@ async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: &ECStore, subsy
for (i, count) in set_drive_counts.iter().enumerate() { for (i, count) in set_drive_counts.iter().enumerate() {
match storageclass::lookup_config(&kvs, *count) { match storageclass::lookup_config(&kvs, *count) {
Ok(res) => { Ok(res) => {
if i == 0 { if i == 0 && GLOBAL_StorageClass.get().is_none() {
if GLOBAL_StorageClass.get().is_none() { if let Err(r) = GLOBAL_StorageClass.set(res) {
if let Err(r) = GLOBAL_StorageClass.set(res) { error!("GLOBAL_StorageClass.set failed {:?}", r);
error!("GLOBAL_StorageClass.set failed {:?}", r);
}
} }
} }
} }
ecstore/src/config/mod.rs
+3 -5
View File
@@ -92,11 +92,9 @@ impl Config {
let mut default = HashMap::new(); let mut default = HashMap::new();
default.insert("_".to_owned(), v.clone()); default.insert("_".to_owned(), v.clone());
self.0.insert(k.clone(), default); self.0.insert(k.clone(), default);
} else { } else if !self.0[k].contains_key("_") {
if !self.0[k].contains_key("_") { if let Some(m) = self.0.get_mut(k) {
if let Some(m) = self.0.get_mut(k) { m.insert("_".to_owned(), v.clone());
m.insert("_".to_owned(), v.clone());
}
} }
} }
} }
ecstore/src/config/storageclass.rs
+2 -4
View File
@@ -307,10 +307,8 @@ pub fn validate_parity_inner(ss_parity: usize, rrs_parity: usize, set_drive_coun
} }
} }
if ss_parity > 0 && rrs_parity > 0 { if ss_parity > 0 && rrs_parity > 0 && ss_parity < rrs_parity {
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)));
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(()) Ok(())
} }
ecstore/src/disk/mod.rs
+295 -4
View File
@@ -21,9 +21,11 @@ use crate::{
}; };
use endpoint::Endpoint; use endpoint::Endpoint;
use futures::StreamExt; use futures::StreamExt;
use local::LocalDisk;
use protos::proto_gen::node_service::{ use protos::proto_gen::node_service::{
node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse, node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse,
}; };
use remote::RemoteDisk;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use std::{ use std::{
any::Any, any::Any,
@@ -44,18 +46,307 @@ use tokio::{
use tokio_stream::wrappers::ReceiverStream; use tokio_stream::wrappers::ReceiverStream;
use tonic::{service::interceptor::InterceptedService, transport::Channel, Request, Status, Streaming}; use tonic::{service::interceptor::InterceptedService, transport::Channel, Request, Status, Streaming};
use tracing::info; use tracing::info;
use tracing::{error, warn}; use tracing::warn;
use uuid::Uuid; use uuid::Uuid;
pub type DiskStore = Arc<Box<dyn DiskAPI>>; pub type DiskStore = Arc<Disk>;
#[derive(Debug)]
pub enum Disk {
Local(LocalDisk),
Remote(RemoteDisk),
}
#[async_trait::async_trait]
impl DiskAPI for Disk {
fn to_string(&self) -> String {
match self {
Disk::Local(local_disk) => local_disk.to_string(),
Disk::Remote(remote_disk) => remote_disk.to_string(),
}
}
fn is_local(&self) -> bool {
match self {
Disk::Local(local_disk) => local_disk.is_local(),
Disk::Remote(remote_disk) => remote_disk.is_local(),
}
}
fn host_name(&self) -> String {
match self {
Disk::Local(local_disk) => local_disk.host_name(),
Disk::Remote(remote_disk) => remote_disk.host_name(),
}
}
async fn is_online(&self) -> bool {
match self {
Disk::Local(local_disk) => local_disk.is_online().await,
Disk::Remote(remote_disk) => remote_disk.is_online().await,
}
}
fn endpoint(&self) -> Endpoint {
match self {
Disk::Local(local_disk) => local_disk.endpoint(),
Disk::Remote(remote_disk) => remote_disk.endpoint(),
}
}
async fn close(&self) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.close().await,
Disk::Remote(remote_disk) => remote_disk.close().await,
}
}
fn path(&self) -> PathBuf {
match self {
Disk::Local(local_disk) => local_disk.path(),
Disk::Remote(remote_disk) => remote_disk.path(),
}
}
fn get_disk_location(&self) -> DiskLocation {
match self {
Disk::Local(local_disk) => local_disk.get_disk_location(),
Disk::Remote(remote_disk) => remote_disk.get_disk_location(),
}
}
async fn get_disk_id(&self) -> Result<Option<Uuid>> {
match self {
Disk::Local(local_disk) => local_disk.get_disk_id().await,
Disk::Remote(remote_disk) => remote_disk.get_disk_id().await,
}
}
async fn set_disk_id(&self, id: Option<Uuid>) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.set_disk_id(id).await,
Disk::Remote(remote_disk) => remote_disk.set_disk_id(id).await,
}
}
async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
match self {
Disk::Local(local_disk) => local_disk.read_all(volume, path).await,
Disk::Remote(remote_disk) => remote_disk.read_all(volume, path).await,
}
}
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await,
Disk::Remote(remote_disk) => remote_disk.write_all(volume, path, data).await,
}
}
async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await,
Disk::Remote(remote_disk) => remote_disk.delete(volume, path, opt).await,
}
}
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
match self {
Disk::Local(local_disk) => local_disk.verify_file(volume, path, fi).await,
Disk::Remote(remote_disk) => remote_disk.verify_file(volume, path, fi).await,
}
}
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
match self {
Disk::Local(local_disk) => local_disk.check_parts(volume, path, fi).await,
Disk::Remote(remote_disk) => remote_disk.check_parts(volume, path, fi).await,
}
}
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await,
Disk::Remote(remote_disk) => {
remote_disk
.rename_part(src_volume, src_path, dst_volume, dst_path, meta)
.await
}
}
}
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await,
Disk::Remote(remote_disk) => remote_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await,
}
}
async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
match self {
Disk::Local(local_disk) => local_disk.create_file(_origvolume, volume, path, _file_size).await,
Disk::Remote(remote_disk) => remote_disk.create_file(_origvolume, volume, path, _file_size).await,
}
}
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
match self {
Disk::Local(local_disk) => local_disk.append_file(volume, path).await,
Disk::Remote(remote_disk) => remote_disk.append_file(volume, path).await,
}
}
async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader> {
match self {
Disk::Local(local_disk) => local_disk.read_file(volume, path).await,
Disk::Remote(remote_disk) => remote_disk.read_file(volume, path).await,
}
}
async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {
match self {
Disk::Local(local_disk) => local_disk.list_dir(_origvolume, volume, _dir_path, _count).await,
Disk::Remote(remote_disk) => remote_disk.list_dir(_origvolume, volume, _dir_path, _count).await,
}
}
async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
match self {
Disk::Local(local_disk) => local_disk.walk_dir(opts).await,
Disk::Remote(remote_disk) => remote_disk.walk_dir(opts).await,
}
}
async fn rename_data(
&self,
src_volume: &str,
src_path: &str,
fi: FileInfo,
dst_volume: &str,
dst_path: &str,
) -> Result<RenameDataResp> {
match self {
Disk::Local(local_disk) => local_disk.rename_data(src_volume, src_path, fi, dst_volume, dst_path).await,
Disk::Remote(remote_disk) => remote_disk.rename_data(src_volume, src_path, fi, dst_volume, dst_path).await,
}
}
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.make_volumes(volumes).await,
Disk::Remote(remote_disk) => remote_disk.make_volumes(volumes).await,
}
}
async fn make_volume(&self, volume: &str) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.make_volume(volume).await,
Disk::Remote(remote_disk) => remote_disk.make_volume(volume).await,
}
}
async fn list_volumes(&self) -> Result<Vec<VolumeInfo>> {
match self {
Disk::Local(local_disk) => local_disk.list_volumes().await,
Disk::Remote(remote_disk) => remote_disk.list_volumes().await,
}
}
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo> {
match self {
Disk::Local(local_disk) => local_disk.stat_volume(volume).await,
Disk::Remote(remote_disk) => remote_disk.stat_volume(volume).await,
}
}
async fn delete_paths(&self, volume: &str, paths: &[&str]) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.delete_paths(volume, paths).await,
Disk::Remote(remote_disk) => remote_disk.delete_paths(volume, paths).await,
}
}
async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.update_metadata(volume, path, fi, opts).await,
Disk::Remote(remote_disk) => remote_disk.update_metadata(volume, path, fi, opts).await,
}
}
async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.write_metadata(_org_volume, volume, path, fi).await,
Disk::Remote(remote_disk) => remote_disk.write_metadata(_org_volume, volume, path, fi).await,
}
}
async fn read_version(
&self,
_org_volume: &str,
volume: &str,
path: &str,
version_id: &str,
opts: &ReadOptions,
) -> Result<FileInfo> {
match self {
Disk::Local(local_disk) => local_disk.read_version(_org_volume, volume, path, version_id, opts).await,
Disk::Remote(remote_disk) => remote_disk.read_version(_org_volume, volume, path, version_id, opts).await,
}
}
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo> {
match self {
Disk::Local(local_disk) => local_disk.read_xl(volume, path, read_data).await,
Disk::Remote(remote_disk) => remote_disk.read_xl(volume, path, read_data).await,
}
}
async fn delete_version(
&self,
volume: &str,
path: &str,
fi: FileInfo,
force_del_marker: bool,
opts: DeleteOptions,
) -> Result<RawFileInfo> {
match self {
Disk::Local(local_disk) => local_disk.delete_version(volume, path, fi, force_del_marker, opts).await,
Disk::Remote(remote_disk) => remote_disk.delete_version(volume, path, fi, force_del_marker, opts).await,
}
}
async fn delete_versions(
&self,
volume: &str,
versions: Vec<FileInfoVersions>,
opts: DeleteOptions,
) -> Result<Vec<Option<Error>>> {
match self {
Disk::Local(local_disk) => local_disk.delete_versions(volume, versions, opts).await,
Disk::Remote(remote_disk) => remote_disk.delete_versions(volume, versions, opts).await,
}
}
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
match self {
Disk::Local(local_disk) => local_disk.read_multiple(req).await,
Disk::Remote(remote_disk) => remote_disk.read_multiple(req).await,
}
}
async fn delete_volume(&self, volume: &str) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.delete_volume(volume).await,
Disk::Remote(remote_disk) => remote_disk.delete_volume(volume).await,
}
}
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo> {
match self {
Disk::Local(local_disk) => local_disk.disk_info(opts).await,
Disk::Remote(remote_disk) => remote_disk.disk_info(opts).await,
}
}
}
pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result<DiskStore> { pub async fn new_disk(ep: &endpoint::Endpoint, opt: &DiskOption) -> Result<DiskStore> {
if ep.is_local { if ep.is_local {
let s = local::LocalDisk::new(ep, opt.cleanup).await?; let s = local::LocalDisk::new(ep, opt.cleanup).await?;
Ok(Arc::new(Box::new(s))) Ok(Arc::new(Disk::Local(s)))
} else { } else {
let remote_disk = remote::RemoteDisk::new(ep, opt).await?; let remote_disk = remote::RemoteDisk::new(ep, opt).await?;
Ok(Arc::new(Box::new(remote_disk))) Ok(Arc::new(Disk::Remote(remote_disk)))
} }
} }
ecstore/src/erasure.rs
+132 -67
View File
@@ -1,4 +1,4 @@
use crate::bitrot::{close_bitrot_writers, BitrotReader, BitrotWriter}; use crate::bitrot::{BitrotReader, BitrotWriter};
use crate::error::{Error, Result, StdError}; use crate::error::{Error, Result, StdError};
use crate::quorum::{object_op_ignored_errs, reduce_write_quorum_errs}; use crate::quorum::{object_op_ignored_errs, reduce_write_quorum_errs};
use bytes::Bytes; use bytes::Bytes;
@@ -7,15 +7,16 @@ use futures::{pin_mut, Stream, StreamExt};
use reed_solomon_erasure::galois_8::ReedSolomon; use reed_solomon_erasure::galois_8::ReedSolomon;
use std::any::Any; use std::any::Any;
use std::fmt::Debug; use std::fmt::Debug;
use tokio::io::AsyncWriteExt; use std::io::ErrorKind;
use tokio::io::DuplexStream; use tokio::io::DuplexStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tracing::debug; use tracing::debug;
use tracing::warn; use tracing::warn;
// use tracing::debug; // use tracing::debug;
use uuid::Uuid; use uuid::Uuid;
// use reader::reader::ChunkedStream; // use reader::reader::ChunkedStream;
use crate::chunk_stream::ChunkedStream; // use crate::chunk_stream::ChunkedStream;
use crate::disk::error::DiskError; use crate::disk::error::DiskError;
pub struct Erasure { pub struct Erasure {
@@ -24,6 +25,7 @@ pub struct Erasure {
encoder: Option<ReedSolomon>, encoder: Option<ReedSolomon>,
pub block_size: usize, pub block_size: usize,
_id: Uuid, _id: Uuid,
buf: Vec<u8>,
} }
impl Erasure { impl Erasure {
@@ -43,12 +45,13 @@ impl Erasure {
block_size, block_size,
encoder, encoder,
_id: Uuid::new_v4(), _id: Uuid::new_v4(),
buf: vec![0u8; block_size],
} }
} }
#[tracing::instrument(level = "debug", skip(self, body, writers))] #[tracing::instrument(level = "debug", skip(self, body, writers))]
pub async fn encode<S>( pub async fn encode<S>(
&self, &mut self,
body: S, body: S,
writers: &mut [Option<BitrotWriter>], writers: &mut [Option<BitrotWriter>],
// block_size: usize, // block_size: usize,
@@ -58,76 +61,138 @@ impl Erasure {
where where
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync, S: Stream<Item = Result<Bytes, StdError>> + Send + Sync,
{ {
// let stream = ChunkedStream::new(body, self.block_size); pin_mut!(body);
let stream = ChunkedStream::new(body, total_size, self.block_size, false); let mut reader = tokio_util::io::StreamReader::new(
body.map(|f| f.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))),
);
let mut total: usize = 0; let mut total: usize = 0;
// let mut idx = 0;
pin_mut!(stream);
// warn!("encode start...");
loop { loop {
match stream.next().await { let new_len = {
Some(result) => match result { let remain = total_size - total;
Ok(data) => { if remain > self.block_size {
total += data.len(); self.block_size
} else {
// EOF remain
if data.is_empty() {
break;
}
// idx += 1;
// warn!("encode {} get data {:?}", data.len(), data.to_vec());
let blocks = self.encode_data(data.as_ref())?;
// warn!(
// "encode shard size: {}/{} from block_size {}, total_size {} ",
// blocks[0].len(),
// blocks.len(),
// data.len(),
// total_size
// );
let mut errs = Vec::new();
for (i, w_op) in writers.iter_mut().enumerate() {
if let Some(w) = w_op {
match w.write(blocks[i].as_ref()).await {
Ok(_) => errs.push(None),
Err(e) => errs.push(Some(e)),
}
} else {
errs.push(Some(Error::new(DiskError::DiskNotFound)));
}
}
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;
} }
};
if new_len == 0 {
break;
}
self.buf.resize(new_len, 0u8);
match reader.read_exact(&mut self.buf).await {
Ok(res) => res,
Err(e) => {
if let ErrorKind::UnexpectedEof = e.kind() {
break;
} else {
return Err(Error::new(e));
}
}
};
total += self.buf.len();
let blocks = self.encode_data(&self.buf)?;
let mut errs = Vec::new();
for (i, w_op) in writers.iter_mut().enumerate() {
if let Some(w) = w_op {
match w.write(blocks[i].as_ref()).await {
Ok(_) => errs.push(None),
Err(e) => errs.push(Some(e)),
}
} else {
errs.push(Some(Error::new(DiskError::DiskNotFound)));
}
}
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);
} }
} }
let _ = close_bitrot_writers(writers).await?;
Ok(total) Ok(total)
// // let stream = ChunkedStream::new(body, self.block_size);
// let stream = ChunkedStream::new(body, total_size, self.block_size, false);
// let mut total: usize = 0;
// // let mut idx = 0;
// pin_mut!(stream);
// // warn!("encode start...");
// loop {
// match stream.next().await {
// Some(result) => match result {
// Ok(data) => {
// total += data.len();
// // EOF
// if data.is_empty() {
// break;
// }
// // idx += 1;
// // warn!("encode {} get data {:?}", data.len(), data.to_vec());
// let blocks = self.encode_data(data.as_ref())?;
// // warn!(
// // "encode shard size: {}/{} from block_size {}, total_size {} ",
// // blocks[0].len(),
// // blocks.len(),
// // data.len(),
// // total_size
// // );
// let mut errs = Vec::new();
// for (i, w_op) in writers.iter_mut().enumerate() {
// if let Some(w) = w_op {
// match w.write(blocks[i].as_ref()).await {
// Ok(_) => errs.push(None),
// Err(e) => errs.push(Some(e)),
// }
// } else {
// errs.push(Some(Error::new(DiskError::DiskNotFound)));
// }
// }
// 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;
// }
// }
// }
// let _ = close_bitrot_writers(writers).await?;
// Ok(total)
} }
pub async fn decode( pub async fn decode(
ecstore/src/file_meta_inline.rs
+1 -1
View File
@@ -16,7 +16,7 @@ impl InlineData {
self.0 = buf.to_vec() self.0 = buf.to_vec()
} }
pub fn as_slice(&self) -> &[u8] { pub fn as_slice(&self) -> &[u8] {
&self.0.as_slice() self.0.as_slice()
} }
pub fn version_ok(&self) -> bool { pub fn version_ok(&self) -> bool {
if self.0.is_empty() { if self.0.is_empty() {
ecstore/src/heal/heal_commands.rs
+1 -1
View File
@@ -7,7 +7,7 @@ use serde::{Deserialize, Serialize};
use tokio::sync::RwLock; use tokio::sync::RwLock;
use crate::{ use crate::{
disk::{DeleteOptions, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}, disk::{DeleteOptions, DiskAPI, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
error::{Error, Result}, error::{Error, Result},
heal::heal_ops::HEALING_TRACKER_FILENAME, heal::heal_ops::HEALING_TRACKER_FILENAME,
new_object_layer_fn, new_object_layer_fn,
ecstore/src/heal/heal_ops.rs
+4 -5
View File
@@ -325,12 +325,12 @@ pub async fn heal_sequence_start(h: Arc<HealSequence>) {
match err { match err {
Some(err) => { Some(err) => {
let mut current_status_w = h.current_status.write().await; let mut current_status_w = h.current_status.write().await;
(*current_status_w).summary = HEAL_STOPPED_STATUS.to_string(); (current_status_w).summary = HEAL_STOPPED_STATUS.to_string();
(*current_status_w).failure_detail = err.to_string(); (current_status_w).failure_detail = err.to_string();
}, },
None => { None => {
let mut current_status_w = h.current_status.write().await; let mut current_status_w = h.current_status.write().await;
(*current_status_w).summary = HEAL_FINISHED_STATUS.to_string(); (current_status_w).summary = HEAL_FINISHED_STATUS.to_string();
} }
} }
}, },
@@ -529,7 +529,7 @@ impl AllHealState {
let _ = self.mu.write().await; let _ = self.mu.write().await;
for (k, v) in self.heal_seq_map.iter() { for (k, v) in self.heal_seq_map.iter() {
if !v.has_ended().await && (has_profix(&k, path_s) || has_profix(path_s, &k)) { if !v.has_ended().await && (has_profix(k, path_s) || has_profix(path_s, k)) {
return Err(Error::from_string(format!( return Err(Error::from_string(format!(
"The provided heal sequence path overlaps with an existing heal path: {}", "The provided heal sequence path overlaps with an existing heal path: {}",
k k
@@ -546,7 +546,6 @@ impl AllHealState {
if heal_sequence.client_token == BG_HEALING_UUID { if heal_sequence.client_token == BG_HEALING_UUID {
// For background heal do nothing, do not spawn an unnecessary goroutine. // For background heal do nothing, do not spawn an unnecessary goroutine.
} else {
} }
todo!() todo!()
} }
ecstore/src/peer.rs
+1
View File
@@ -7,6 +7,7 @@ use std::{collections::HashMap, fmt::Debug, sync::Arc};
use tonic::Request; use tonic::Request;
use tracing::warn; use tracing::warn;
use crate::disk::DiskAPI;
use crate::store::all_local_disk; use crate::store::all_local_disk;
use crate::{ use crate::{
disk::{self, error::DiskError, VolumeInfo}, disk::{self, error::DiskError, VolumeInfo},
ecstore/src/sets.rs
+2 -2
View File
@@ -11,7 +11,7 @@ use uuid::Uuid;
use crate::{ use crate::{
disk::{ disk::{
format::{DistributionAlgoVersion, FormatV3}, format::{DistributionAlgoVersion, FormatV3},
DiskStore, DiskAPI, DiskStore,
}, },
endpoints::PoolEndpoints, endpoints::PoolEndpoints,
error::{Error, Result}, error::{Error, Result},
@@ -454,7 +454,7 @@ impl StorageAPI for Sets {
delimiter: &str, delimiter: &str,
max_uploads: usize, max_uploads: usize,
) -> Result<ListMultipartsInfo> { ) -> Result<ListMultipartsInfo> {
self.get_disks_by_key(&prefix) self.get_disks_by_key(prefix)
.list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads) .list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads)
.await .await
} }
ecstore/src/store.rs
+33 -45
View File
@@ -5,7 +5,7 @@ use crate::bucket::metadata_sys::{self, init_bucket_metadata_sys, set_bucket_met
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname}; use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
use crate::config::{self, storageclass, GLOBAL_ConfigSys}; use crate::config::{self, storageclass, GLOBAL_ConfigSys};
use crate::disk::endpoint::EndpointType; use crate::disk::endpoint::EndpointType;
use crate::disk::{DiskInfo, DiskInfoOptions, MetaCacheEntry}; use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions, MetaCacheEntry};
use crate::global::{ use crate::global::{
is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, DISK_ASSUME_UNKNOWN_SIZE, DISK_FILL_FRACTION, is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, DISK_ASSUME_UNKNOWN_SIZE, DISK_FILL_FRACTION,
DISK_MIN_INODES, DISK_RESERVE_FRACTION, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, DISK_MIN_INODES, DISK_RESERVE_FRACTION, GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES,
@@ -331,7 +331,7 @@ impl ECStore {
async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> { async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option<usize> {
let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await; let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await;
server_pools.filter_max_used(100 - (100 as f64 * DISK_RESERVE_FRACTION) as u64); server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64);
let total = server_pools.total_available(); let total = server_pools.total_available();
if total == 0 { if total == 0 {
@@ -382,10 +382,7 @@ impl ECStore {
} }
if !is_meta_bucketname(bucket) { if !is_meta_bucketname(bucket) {
let avail = match has_space_for(zinfo, size).await { let avail = has_space_for(zinfo, size).await.unwrap_or_default();
Ok(res) => res,
Err(_err) => false,
};
if !avail { if !avail {
server_pools.push(PoolAvailableSpace { server_pools.push(PoolAvailableSpace {
@@ -399,19 +396,17 @@ impl ECStore {
let mut available = 0; let mut available = 0;
let mut max_used_pct = 0; let mut max_used_pct = 0;
for disk_op in zinfo.iter() { for disk in zinfo.iter().flatten() {
if let Some(disk) = disk_op { if disk.total == 0 {
if disk.total == 0 { continue;
continue; }
}
available += disk.total - disk.used; available += disk.total - disk.used;
let pct_used = disk.used * 100 / disk.total; let pct_used = disk.used * 100 / disk.total;
if pct_used > max_used_pct { if pct_used > max_used_pct {
max_used_pct = pct_used; max_used_pct = pct_used;
}
} }
} }
@@ -769,7 +764,7 @@ impl ObjectIO for ECStore {
return self.pools[0].put_object(bucket, object.as_str(), data, opts).await; return self.pools[0].put_object(bucket, object.as_str(), data, opts).await;
} }
let idx = self.get_pool_idx(&bucket, &object, data.content_length as i64).await?; let idx = self.get_pool_idx(bucket, &object, data.content_length as i64).await?;
if opts.data_movement && idx == opts.src_pool_idx { if opts.data_movement && idx == opts.src_pool_idx {
return Err(Error::new(StorageError::DataMovementOverwriteErr( return Err(Error::new(StorageError::DataMovementOverwriteErr(
@@ -812,11 +807,11 @@ impl StorageAPI for ECStore {
} }
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> { async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
if is_meta_bucketname(&bucket) { if is_meta_bucketname(bucket) {
return Err(StorageError::BucketNameInvalid(bucket.to_string()).into()); return Err(StorageError::BucketNameInvalid(bucket.to_string()).into());
} }
if let Err(err) = check_valid_bucket_name(&bucket) { if let Err(err) = check_valid_bucket_name(bucket) {
return Err(StorageError::BucketNameInvalid(err.to_string()).into()); return Err(StorageError::BucketNameInvalid(err.to_string()).into());
} }
@@ -838,8 +833,8 @@ impl StorageAPI for ECStore {
Ok(()) Ok(())
} }
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
if !is_meta_bucketname(&bucket) { if !is_meta_bucketname(bucket) {
if let Err(err) = check_valid_bucket_name_strict(&bucket) { if let Err(err) = check_valid_bucket_name_strict(bucket) {
return Err(StorageError::BucketNameInvalid(err.to_string()).into()); return Err(StorageError::BucketNameInvalid(err.to_string()).into());
} }
@@ -885,7 +880,7 @@ impl StorageAPI for ECStore {
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> { async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?; let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?;
if let Ok(sys) = metadata_sys::get(&bucket).await { if let Ok(sys) = metadata_sys::get(bucket).await {
info.created = Some(sys.created); info.created = Some(sys.created);
info.versionning = sys.versioning(); info.versionning = sys.versioning();
info.object_locking = sys.object_locking(); info.object_locking = sys.object_locking();
@@ -1234,7 +1229,7 @@ impl StorageAPI for ECStore {
for (idx, pool) in self.pools.iter().enumerate() { for (idx, pool) in self.pools.iter().enumerate() {
// // TODO: IsSuspended // // TODO: IsSuspended
let res = pool let res = pool
.list_multipart_uploads(bucket, &object, "", "", "", MAX_UPLOADS_LIST) .list_multipart_uploads(bucket, object, "", "", "", MAX_UPLOADS_LIST)
.await?; .await?;
if !res.uploads.is_empty() { if !res.uploads.is_empty() {
@@ -1376,7 +1371,7 @@ impl StorageAPI for ECStore {
// Return the first nil error // Return the first nil error
for i in 0..self.pools.len() { for i in 0..self.pools.len() {
if errs.get(&i).is_none() { if !errs.contains_key(&i) {
return Ok(results.remove(&i).unwrap()); return Ok(results.remove(&i).unwrap());
} }
} }
@@ -1385,7 +1380,7 @@ impl StorageAPI for ECStore {
for (k, err) in errs.iter() { for (k, err) in errs.iter() {
match err.downcast_ref::<DiskError>() { match err.downcast_ref::<DiskError>() {
Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {} Some(DiskError::FileNotFound) | Some(DiskError::FileVersionNotFound) => {}
_ => return Ok(results.remove(&k).unwrap()), _ => return Ok(results.remove(k).unwrap()),
} }
} }
@@ -1611,10 +1606,7 @@ fn check_multipart_object_args(bucket: &str, object: &str, upload_id: &str) -> R
if let Err(e) = base64_decode(upload_id.as_bytes()) { if let Err(e) = base64_decode(upload_id.as_bytes()) {
return Err(Error::new(StorageError::MalformedUploadID(format!( return Err(Error::new(StorageError::MalformedUploadID(format!(
"{}/{}-{},err:{}", "{}/{}-{},err:{}",
bucket, bucket, object, upload_id, e
object,
upload_id,
e.to_string()
)))); ))));
}; };
check_object_args(bucket, object) check_object_args(bucket, object)
@@ -1711,7 +1703,7 @@ pub async fn heal_entry(
Ok(()) Ok(())
} }
async fn get_disk_infos(disks: &Vec<Option<DiskStore>>) -> Vec<Option<DiskInfo>> { async fn get_disk_infos(disks: &[Option<DiskStore>]) -> Vec<Option<DiskInfo>> {
let opts = &DiskInfoOptions::default(); let opts = &DiskInfoOptions::default();
let mut res = vec![None; disks.len()]; let mut res = vec![None; disks.len()];
for (idx, disk_op) in disks.iter().enumerate() { for (idx, disk_op) in disks.iter().enumerate() {
@@ -1776,7 +1768,7 @@ impl ServerPoolsAvailableSpace {
} }
} }
async fn has_space_for(dis: &Vec<Option<DiskInfo>>, size: i64) -> Result<bool> { async fn has_space_for(dis: &[Option<DiskInfo>], size: i64) -> Result<bool> {
let size = { let size = {
if size < 0 { if size < 0 {
DISK_ASSUME_UNKNOWN_SIZE DISK_ASSUME_UNKNOWN_SIZE
@@ -1789,12 +1781,10 @@ async fn has_space_for(dis: &Vec<Option<DiskInfo>>, size: i64) -> Result<bool> {
let mut total = 0; let mut total = 0;
let mut disks_num = 0; let mut disks_num = 0;
for disk_op in dis.iter() { for disk in dis.iter().flatten() {
if let Some(disk) = disk_op { disks_num += 1;
disks_num += 1; total += disk.total;
total += disk.total; available += disk.total - disk.used;
available += disk.total - disk.used;
}
} }
if disks_num < dis.len() / 2 || disks_num == 0 { if disks_num < dis.len() / 2 || disks_num == 0 {
@@ -1807,15 +1797,13 @@ async fn has_space_for(dis: &Vec<Option<DiskInfo>>, size: i64) -> Result<bool> {
let per_disk = size / disks_num as u64; let per_disk = size / disks_num as u64;
for disk_op in dis.iter() { for disk in dis.iter().flatten() {
if let Some(disk) = disk_op { if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 {
if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 { return Ok(false);
return Ok(false); }
}
if disk.free <= per_disk { if disk.free <= per_disk {
return Ok(false); return Ok(false);
}
} }
} }
ecstore/src/store_api.rs
+3 -21
View File
@@ -172,27 +172,9 @@ impl FileInfo {
let (content_type, content_encoding, etag) = { let (content_type, content_encoding, etag) = {
if let Some(ref meta) = self.metadata { if let Some(ref meta) = self.metadata {
let content_type = { let content_type = meta.get("content-type").cloned();
if let Some(ty) = meta.get("content-type") { let content_encoding = meta.get("content-encoding").cloned();
Some(ty.clone()) let etag = meta.get("etag").cloned();
} 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) (content_type, content_encoding, etag)
} else { } else {
ecstore/src/store_err.rs
+4 -16
View File
@@ -42,10 +42,7 @@ pub enum StorageError {
pub fn is_err_invalid_upload_id(err: &Error) -> bool { pub fn is_err_invalid_upload_id(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() { if let Some(e) = err.downcast_ref::<StorageError>() {
match e { matches!(e, StorageError::InvalidUploadID(_, _, _))
StorageError::InvalidUploadID(_, _, _) => true,
_ => false,
}
} else { } else {
false false
} }
@@ -53,10 +50,7 @@ pub fn is_err_invalid_upload_id(err: &Error) -> bool {
pub fn is_err_version_not_found(err: &Error) -> bool { pub fn is_err_version_not_found(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() { if let Some(e) = err.downcast_ref::<StorageError>() {
match e { matches!(e, StorageError::VersionNotFound(_, _, _))
StorageError::VersionNotFound(_, _, _) => true,
_ => false,
}
} else { } else {
false false
} }
@@ -64,10 +58,7 @@ pub fn is_err_version_not_found(err: &Error) -> bool {
pub fn is_err_bucket_exists(err: &Error) -> bool { pub fn is_err_bucket_exists(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() { if let Some(e) = err.downcast_ref::<StorageError>() {
match e { matches!(e, StorageError::BucketExists(_))
StorageError::BucketExists(_) => true,
_ => false,
}
} else { } else {
false false
} }
@@ -78,10 +69,7 @@ pub fn is_err_object_not_found(err: &Error) -> bool {
return true; return true;
} }
if let Some(e) = err.downcast_ref::<StorageError>() { if let Some(e) = err.downcast_ref::<StorageError>() {
match e { matches!(e, StorageError::ObjectNotFound(_, _))
StorageError::ObjectNotFound(_, _) => true,
_ => false,
}
} else { } else {
false false
} }
ecstore/src/store_init.rs
+1
View File
@@ -1,4 +1,5 @@
use crate::config::{storageclass, KVS}; use crate::config::{storageclass, KVS};
use crate::disk::DiskAPI;
use crate::{ use crate::{
disk::{ disk::{
error::DiskError, error::DiskError,
reader/src/hasher.rs
+11 -3
View File
@@ -78,6 +78,11 @@ impl Sha256 {
} }
} }
} }
impl Default for Sha256 {
fn default() -> Self {
Self::new()
}
}
impl Hasher for Sha256 { impl Hasher for Sha256 {
fn write(&mut self, bytes: &[u8]) { fn write(&mut self, bytes: &[u8]) {
@@ -110,6 +115,11 @@ impl MD5 {
Self { hasher: Md5::new() } Self { hasher: Md5::new() }
} }
} }
impl Default for MD5 {
fn default() -> Self {
Self::new()
}
}
impl Hasher for MD5 { impl Hasher for MD5 {
fn write(&mut self, bytes: &[u8]) { fn write(&mut self, bytes: &[u8]) {
@@ -142,9 +152,7 @@ impl Uuid {
} }
impl Hasher for Uuid { impl Hasher for Uuid {
fn write(&mut self, _bytes: &[u8]) { fn write(&mut self, _bytes: &[u8]) {}
()
}
fn reset(&mut self) {} fn reset(&mut self) {}
reader/src/reader.rs
+7 -8
View File
@@ -1,7 +1,6 @@
use bytes::Bytes; use bytes::Bytes;
use s3s::StdError; use s3s::StdError;
use std::collections::VecDeque; use std::collections::VecDeque;
use tracing::warn;
use std::pin::Pin; use std::pin::Pin;
use std::task::Poll; use std::task::Poll;
@@ -145,7 +144,7 @@ where
} }
if let Some(md5) = this.md5 { if let Some(md5) = this.md5 {
let _ = md5.write(bytes); md5.write(bytes);
} }
} }
Some(Err(err)) => { Some(Err(err)) => {
@@ -231,7 +230,7 @@ where
// println!("get len {}", bytes.len()); // println!("get len {}", bytes.len());
// 如果有剩余 // 如果有剩余
if this.remaining.len() > 0 { if !this.remaining.is_empty() {
let need_size = chuck_size - this.remaining.len(); let need_size = chuck_size - this.remaining.len();
// 传入的数据大小需要补齐的大小，使用传入数据补齐 // 传入的数据大小需要补齐的大小，使用传入数据补齐
if bytes.len() >= need_size { if bytes.len() >= need_size {
@@ -255,7 +254,7 @@ where
this.streams.push_back(chuck); this.streams.push_back(chuck);
} }
if bytes.len() > 0 { if !bytes.is_empty() {
this.remaining.extend_from_slice(&bytes); this.remaining.extend_from_slice(&bytes);
} }
@@ -267,21 +266,21 @@ where
return Poll::Pending; return Poll::Pending;
} }
if this.remaining.len() > 0 { if !this.remaining.is_empty() {
let b = this.remaining.clone(); let b = this.remaining.clone();
this.remaining.clear(); this.remaining.clear();
return Poll::Ready(Some(Ok(Bytes::from(b)))); return Poll::Ready(Some(Ok(Bytes::from(b))));
} }
Poll::Ready(None) Poll::Ready(None)
} }
Err(err) => return Poll::Ready(Some(Err(err))), Err(err) => Poll::Ready(Some(Err(err))),
}, },
None => { None => {
// println!("get empty"); // println!("get empty");
if let Some(b) = this.streams.pop_front() { if let Some(b) = this.streams.pop_front() {
return Poll::Ready(Some(Ok(b))); return Poll::Ready(Some(Ok(b)));
} }
if this.remaining.len() > 0 { if !this.remaining.is_empty() {
let b = this.remaining.clone(); let b = this.remaining.clone();
this.remaining.clear(); this.remaining.clear();
return Poll::Ready(Some(Ok(Bytes::from(b)))); return Poll::Ready(Some(Ok(Bytes::from(b))));
@@ -343,7 +342,7 @@ where
fn size_hint(&self) -> (usize, Option<usize>) { fn size_hint(&self) -> (usize, Option<usize>) {
let mut items = self.streams.len(); let mut items = self.streams.len();
if self.remaining.len() > 0 { if !self.remaining.is_empty() {
items += 1; items += 1;
} }
(items, Some(items)) (items, Some(items))
rustfs/Cargo.toml
+1
View File
@@ -35,6 +35,7 @@ s3s.workspace = true
serde_json.workspace = true serde_json.workspace = true
tracing.workspace = true tracing.workspace = true
time = { workspace = true, features = ["parsing", "formatting"] } time = { workspace = true, features = ["parsing", "formatting"] }
tokio-util = { version = "0.7.12", features = ["io", "compat"] }
tokio = { workspace = true, features = [ tokio = { workspace = true, features = [
"rt-multi-thread", "rt-multi-thread",
"macros", "macros",
rustfs/src/grpc.rs
+2 -2
View File
@@ -2,8 +2,8 @@ use std::{error::Error, io::ErrorKind, pin::Pin};
use ecstore::{ use ecstore::{
disk::{ disk::{
DeleteOptions, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, Reader, UpdateMetadataOpts, DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, Reader,
WalkDirOptions, UpdateMetadataOpts, WalkDirOptions,
}, },
erasure::{ReadAt, Writer}, erasure::{ReadAt, Writer},
peer::{LocalPeerS3Client, PeerS3Client}, peer::{LocalPeerS3Client, PeerS3Client},
rustfs/src/storage/acess.rs
+6 -1
View File
@@ -1,3 +1,5 @@
use std::str::FromStr;
use super::ecfs::FS; use super::ecfs::FS;
use ecstore::bucket::policy::action::Action; use ecstore::bucket::policy::action::Action;
@@ -42,7 +44,10 @@ impl S3Access for FS {
// 上层验证了 ak/sk // 上层验证了 ak/sk
// warn!("check s3_op {:?} cred {:?}", cx.s3_op().name(), cx.credentials()); // warn!("check s3_op {:?} cred {:?}", cx.s3_op().name(), cx.credentials());
let action = Action::from_str(format!("s3:{}", cx.s3_op().name()).as_str()); let action = match Action::from_str(format!("s3:{}", cx.s3_op().name()).as_str()) {
Ok(res) => Some(res),
Err(_) => None,
};
let req_info = ReqInfo { let req_info = ReqInfo {
card: cx.credentials().cloned(), card: cx.credentials().cloned(),
scripts/run.sh
+3 -3
View File
@@ -6,9 +6,9 @@ mkdir -p ./target/volume/test
mkdir -p ./target/volume/test{0..4} mkdir -p ./target/volume/test{0..4}
if [ -z "$RUST_LOG" ]; then # if [ -z "$RUST_LOG" ]; then
export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,reader=debug" # export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug,reader=debug"
fi # fi
# export RUSTFS_ERASURE_SET_DRIVE_COUNT=5 # export RUSTFS_ERASURE_SET_DRIVE_COUNT=5