mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 00:38:16 +00:00
cargo clippy
This commit is contained in:
@@ -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 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)> {
|
||||
|
||||
@@ -3,6 +3,7 @@ use lazy_static::lazy_static;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
str::FromStr,
|
||||
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 {
|
||||
"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,
|
||||
"s3:AbortMultipartUpload" => Ok(Action::AbortMultipartUpload),
|
||||
"s3:CreateBucket" => Ok(Action::CreateBucket),
|
||||
"s3:DeleteBucket" => Ok(Action::DeleteBucket),
|
||||
"s3:ForceDeleteBucket" => Ok(Action::ForceDeleteBucket),
|
||||
"s3:DeleteBucketPolicy" => Ok(Action::DeleteBucketPolicy),
|
||||
"s3:DeleteBucketCors" => Ok(Action::DeleteBucketCors),
|
||||
"s3:DeleteObject" => Ok(Action::DeleteObject),
|
||||
"s3:GetBucketLocation" => Ok(Action::GetBucketLocation),
|
||||
"s3:GetBucketNotification" => Ok(Action::GetBucketNotification),
|
||||
"s3:GetBucketPolicy" => Ok(Action::GetBucketPolicy),
|
||||
"s3:GetBucketCors" => Ok(Action::GetBucketCors),
|
||||
"s3:GetObject" => Ok(Action::GetObject),
|
||||
"s3:GetObjectAttributes" => Ok(Action::GetObjectAttributes),
|
||||
"s3:HeadBucket" => Ok(Action::HeadBucket),
|
||||
"s3:ListAllMyBuckets" => Ok(Action::ListAllMyBuckets),
|
||||
"s3:ListBucket" => Ok(Action::ListBucket),
|
||||
"s3:GetBucketPolicyStatus" => Ok(Action::GetBucketPolicyStatus),
|
||||
"s3:ListBucketVersions" => Ok(Action::ListBucketVersions),
|
||||
"s3:ListBucketMultipartUploads" => Ok(Action::ListBucketMultipartUploads),
|
||||
"s3:ListenNotification" => Ok(Action::ListenNotification),
|
||||
"s3:ListenBucketNotification" => Ok(Action::ListenBucketNotification),
|
||||
"s3:ListMultipartUploadParts" => Ok(Action::ListMultipartUploadParts),
|
||||
"s3:PutLifecycleConfiguration" => Ok(Action::PutLifecycleConfiguration),
|
||||
"s3:GetLifecycleConfiguration" => Ok(Action::GetLifecycleConfiguration),
|
||||
"s3:PutBucketNotification" => Ok(Action::PutBucketNotification),
|
||||
"s3:PutBucketPolicy" => Ok(Action::PutBucketPolicy),
|
||||
"s3:PutBucketCors" => Ok(Action::PutBucketCors),
|
||||
"s3:PutObject" => Ok(Action::PutObject),
|
||||
"s3:DeleteObjectVersion" => Ok(Action::DeleteObjectVersion),
|
||||
"s3:DeleteObjectVersionTagging" => Ok(Action::DeleteObjectVersionTagging),
|
||||
"s3:GetObjectVersion" => Ok(Action::GetObjectVersion),
|
||||
"s3:GetObjectVersionAttributes" => Ok(Action::GetObjectVersionAttributes),
|
||||
"s3:GetObjectVersionTagging" => Ok(Action::GetObjectVersionTagging),
|
||||
"s3:PutObjectVersionTagging" => Ok(Action::PutObjectVersionTagging),
|
||||
"s3:BypassGovernanceRetention" => Ok(Action::BypassGovernanceRetention),
|
||||
"s3:PutObjectRetention" => Ok(Action::PutObjectRetention),
|
||||
"s3:GetObjectRetention" => Ok(Action::GetObjectRetention),
|
||||
"s3:GetObjectLegalHold" => Ok(Action::GetObjectLegalHold),
|
||||
"s3:PutObjectLegalHold" => Ok(Action::PutObjectLegalHold),
|
||||
"s3:GetBucketObjectLockConfiguration" => Ok(Action::GetBucketObjectLockConfiguration),
|
||||
"s3:PutBucketObjectLockConfiguration" => Ok(Action::PutBucketObjectLockConfiguration),
|
||||
"s3:GetBucketTagging" => Ok(Action::GetBucketTagging),
|
||||
"s3:PutBucketTagging" => Ok(Action::PutBucketTagging),
|
||||
"s3:GetObjectTagging" => Ok(Action::GetObjectTagging),
|
||||
"s3:PutObjectTagging" => Ok(Action::PutObjectTagging),
|
||||
"s3:DeleteObjectTagging" => Ok(Action::DeleteObjectTagging),
|
||||
"s3:PutEncryptionConfiguration" => Ok(Action::PutBucketEncryption),
|
||||
"s3:GetEncryptionConfiguration" => Ok(Action::GetBucketEncryption),
|
||||
"s3:PutBucketVersioning" => Ok(Action::PutBucketVersioning),
|
||||
"s3:GetBucketVersioning" => Ok(Action::GetBucketVersioning),
|
||||
"s3:PutReplicationConfiguration" => Ok(Action::PutReplicationConfiguration),
|
||||
"s3:GetReplicationConfiguration" => Ok(Action::GetReplicationConfiguration),
|
||||
"s3:ReplicateObject" => Ok(Action::ReplicateObject),
|
||||
"s3:ReplicateDelete" => Ok(Action::ReplicateDelete),
|
||||
"s3:ReplicateTags" => Ok(Action::ReplicateTags),
|
||||
"s3:GetObjectVersionForReplication" => Ok(Action::GetObjectVersionForReplication),
|
||||
"s3:RestoreObject" => Ok(Action::RestoreObject),
|
||||
"s3:ResetBucketReplicationState" => Ok(Action::ResetBucketReplicationState),
|
||||
"s3:PutObjectFanOut" => Ok(Action::PutObjectFanOut),
|
||||
"s3:*" => Ok(Action::AllActions),
|
||||
_ => Err(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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() {
|
||||
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);
|
||||
}
|
||||
if i == 0 && GLOBAL_StorageClass.get().is_none() {
|
||||
if let Err(r) = GLOBAL_StorageClass.set(res) {
|
||||
error!("GLOBAL_StorageClass.set failed {:?}", r);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -92,11 +92,9 @@ impl Config {
|
||||
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());
|
||||
}
|
||||
} else if !self.0[k].contains_key("_") {
|
||||
if let Some(m) = self.0.get_mut(k) {
|
||||
m.insert("_".to_owned(), v.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 < 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)));
|
||||
}
|
||||
if ss_parity > 0 && rrs_parity > 0 && 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(())
|
||||
}
|
||||
|
||||
@@ -44,7 +44,7 @@ use tokio::{
|
||||
use tokio_stream::wrappers::ReceiverStream;
|
||||
use tonic::{service::interceptor::InterceptedService, transport::Channel, Request, Status, Streaming};
|
||||
use tracing::info;
|
||||
use tracing::{error, warn};
|
||||
use tracing::warn;
|
||||
use uuid::Uuid;
|
||||
|
||||
pub type DiskStore = Arc<Box<dyn DiskAPI>>;
|
||||
|
||||
@@ -16,7 +16,7 @@ impl InlineData {
|
||||
self.0 = buf.to_vec()
|
||||
}
|
||||
pub fn as_slice(&self) -> &[u8] {
|
||||
&self.0.as_slice()
|
||||
self.0.as_slice()
|
||||
}
|
||||
pub fn version_ok(&self) -> bool {
|
||||
if self.0.is_empty() {
|
||||
|
||||
@@ -325,12 +325,12 @@ pub async fn heal_sequence_start(h: Arc<HealSequence>) {
|
||||
match err {
|
||||
Some(err) => {
|
||||
let mut current_status_w = h.current_status.write().await;
|
||||
(*current_status_w).summary = HEAL_STOPPED_STATUS.to_string();
|
||||
(*current_status_w).failure_detail = err.to_string();
|
||||
(current_status_w).summary = HEAL_STOPPED_STATUS.to_string();
|
||||
(current_status_w).failure_detail = err.to_string();
|
||||
},
|
||||
None => {
|
||||
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;
|
||||
|
||||
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!(
|
||||
"The provided heal sequence path overlaps with an existing heal path: {}",
|
||||
k
|
||||
@@ -546,7 +546,6 @@ impl AllHealState {
|
||||
|
||||
if heal_sequence.client_token == BG_HEALING_UUID {
|
||||
// For background heal do nothing, do not spawn an unnecessary goroutine.
|
||||
} else {
|
||||
}
|
||||
todo!()
|
||||
}
|
||||
|
||||
+1
-1
@@ -454,7 +454,7 @@ impl StorageAPI for Sets {
|
||||
delimiter: &str,
|
||||
max_uploads: usize,
|
||||
) -> 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)
|
||||
.await
|
||||
}
|
||||
|
||||
+32
-44
@@ -331,7 +331,7 @@ impl ECStore {
|
||||
|
||||
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;
|
||||
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();
|
||||
|
||||
if total == 0 {
|
||||
@@ -382,10 +382,7 @@ impl ECStore {
|
||||
}
|
||||
|
||||
if !is_meta_bucketname(bucket) {
|
||||
let avail = match has_space_for(zinfo, size).await {
|
||||
Ok(res) => res,
|
||||
Err(_err) => false,
|
||||
};
|
||||
let avail = has_space_for(zinfo, size).await.unwrap_or_default();
|
||||
|
||||
if !avail {
|
||||
server_pools.push(PoolAvailableSpace {
|
||||
@@ -399,19 +396,17 @@ impl ECStore {
|
||||
|
||||
let mut available = 0;
|
||||
let mut max_used_pct = 0;
|
||||
for disk_op in zinfo.iter() {
|
||||
if let Some(disk) = disk_op {
|
||||
if disk.total == 0 {
|
||||
continue;
|
||||
}
|
||||
for disk in zinfo.iter().flatten() {
|
||||
if disk.total == 0 {
|
||||
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 {
|
||||
max_used_pct = pct_used;
|
||||
}
|
||||
if pct_used > max_used_pct {
|
||||
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;
|
||||
}
|
||||
|
||||
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 {
|
||||
return Err(Error::new(StorageError::DataMovementOverwriteErr(
|
||||
@@ -812,11 +807,11 @@ impl StorageAPI for ECStore {
|
||||
}
|
||||
|
||||
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());
|
||||
}
|
||||
|
||||
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());
|
||||
}
|
||||
|
||||
@@ -838,8 +833,8 @@ impl StorageAPI for ECStore {
|
||||
Ok(())
|
||||
}
|
||||
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
|
||||
if !is_meta_bucketname(&bucket) {
|
||||
if let Err(err) = check_valid_bucket_name_strict(&bucket) {
|
||||
if !is_meta_bucketname(bucket) {
|
||||
if let Err(err) = check_valid_bucket_name_strict(bucket) {
|
||||
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> {
|
||||
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.versionning = sys.versioning();
|
||||
info.object_locking = sys.object_locking();
|
||||
@@ -1234,7 +1229,7 @@ impl StorageAPI for ECStore {
|
||||
for (idx, pool) in self.pools.iter().enumerate() {
|
||||
// // TODO: IsSuspended
|
||||
let res = pool
|
||||
.list_multipart_uploads(bucket, &object, "", "", "", MAX_UPLOADS_LIST)
|
||||
.list_multipart_uploads(bucket, object, "", "", "", MAX_UPLOADS_LIST)
|
||||
.await?;
|
||||
|
||||
if !res.uploads.is_empty() {
|
||||
@@ -1376,7 +1371,7 @@ impl StorageAPI for ECStore {
|
||||
|
||||
// Return the first nil error
|
||||
for i in 0..self.pools.len() {
|
||||
if errs.get(&i).is_none() {
|
||||
if !errs.contains_key(&i) {
|
||||
return Ok(results.remove(&i).unwrap());
|
||||
}
|
||||
}
|
||||
@@ -1385,7 +1380,7 @@ impl StorageAPI for ECStore {
|
||||
for (k, err) in errs.iter() {
|
||||
match err.downcast_ref::<DiskError>() {
|
||||
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()) {
|
||||
return Err(Error::new(StorageError::MalformedUploadID(format!(
|
||||
"{}/{}-{},err:{}",
|
||||
bucket,
|
||||
object,
|
||||
upload_id,
|
||||
e.to_string()
|
||||
bucket, object, upload_id, e
|
||||
))));
|
||||
};
|
||||
check_object_args(bucket, object)
|
||||
@@ -1711,7 +1703,7 @@ pub async fn heal_entry(
|
||||
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 mut res = vec![None; disks.len()];
|
||||
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 = {
|
||||
if size < 0 {
|
||||
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 disks_num = 0;
|
||||
|
||||
for disk_op in dis.iter() {
|
||||
if let Some(disk) = disk_op {
|
||||
disks_num += 1;
|
||||
total += disk.total;
|
||||
available += disk.total - disk.used;
|
||||
}
|
||||
for disk in dis.iter().flatten() {
|
||||
disks_num += 1;
|
||||
total += disk.total;
|
||||
available += disk.total - disk.used;
|
||||
}
|
||||
|
||||
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;
|
||||
|
||||
for disk_op in dis.iter() {
|
||||
if let Some(disk) = disk_op {
|
||||
if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 {
|
||||
return Ok(false);
|
||||
}
|
||||
for disk in dis.iter().flatten() {
|
||||
if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 {
|
||||
return Ok(false);
|
||||
}
|
||||
|
||||
if disk.free <= per_disk {
|
||||
return Ok(false);
|
||||
}
|
||||
if disk.free <= per_disk {
|
||||
return Ok(false);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -172,27 +172,9 @@ impl FileInfo {
|
||||
|
||||
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
|
||||
}
|
||||
};
|
||||
let content_type = meta.get("content-type").cloned();
|
||||
let content_encoding = meta.get("content-encoding").cloned();
|
||||
let etag = meta.get("etag").cloned();
|
||||
|
||||
(content_type, content_encoding, etag)
|
||||
} else {
|
||||
|
||||
@@ -42,10 +42,7 @@ pub enum StorageError {
|
||||
|
||||
pub fn is_err_invalid_upload_id(err: &Error) -> bool {
|
||||
if let Some(e) = err.downcast_ref::<StorageError>() {
|
||||
match e {
|
||||
StorageError::InvalidUploadID(_, _, _) => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(e, StorageError::InvalidUploadID(_, _, _))
|
||||
} else {
|
||||
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 {
|
||||
if let Some(e) = err.downcast_ref::<StorageError>() {
|
||||
match e {
|
||||
StorageError::VersionNotFound(_, _, _) => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(e, StorageError::VersionNotFound(_, _, _))
|
||||
} else {
|
||||
false
|
||||
}
|
||||
@@ -64,10 +58,7 @@ pub fn is_err_version_not_found(err: &Error) -> bool {
|
||||
|
||||
pub fn is_err_bucket_exists(err: &Error) -> bool {
|
||||
if let Some(e) = err.downcast_ref::<StorageError>() {
|
||||
match e {
|
||||
StorageError::BucketExists(_) => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(e, StorageError::BucketExists(_))
|
||||
} else {
|
||||
false
|
||||
}
|
||||
@@ -78,10 +69,7 @@ pub fn is_err_object_not_found(err: &Error) -> bool {
|
||||
return true;
|
||||
}
|
||||
if let Some(e) = err.downcast_ref::<StorageError>() {
|
||||
match e {
|
||||
StorageError::ObjectNotFound(_, _) => true,
|
||||
_ => false,
|
||||
}
|
||||
matches!(e, StorageError::ObjectNotFound(_, _))
|
||||
} else {
|
||||
false
|
||||
}
|
||||
|
||||
+11
-3
@@ -78,6 +78,11 @@ impl Sha256 {
|
||||
}
|
||||
}
|
||||
}
|
||||
impl Default for Sha256 {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Hasher for Sha256 {
|
||||
fn write(&mut self, bytes: &[u8]) {
|
||||
@@ -110,6 +115,11 @@ impl MD5 {
|
||||
Self { hasher: Md5::new() }
|
||||
}
|
||||
}
|
||||
impl Default for MD5 {
|
||||
fn default() -> Self {
|
||||
Self::new()
|
||||
}
|
||||
}
|
||||
|
||||
impl Hasher for MD5 {
|
||||
fn write(&mut self, bytes: &[u8]) {
|
||||
@@ -142,9 +152,7 @@ impl Uuid {
|
||||
}
|
||||
|
||||
impl Hasher for Uuid {
|
||||
fn write(&mut self, _bytes: &[u8]) {
|
||||
()
|
||||
}
|
||||
fn write(&mut self, _bytes: &[u8]) {}
|
||||
|
||||
fn reset(&mut self) {}
|
||||
|
||||
|
||||
@@ -1,7 +1,6 @@
|
||||
use bytes::Bytes;
|
||||
use s3s::StdError;
|
||||
use std::collections::VecDeque;
|
||||
use tracing::warn;
|
||||
|
||||
use std::pin::Pin;
|
||||
use std::task::Poll;
|
||||
@@ -145,7 +144,7 @@ where
|
||||
}
|
||||
|
||||
if let Some(md5) = this.md5 {
|
||||
let _ = md5.write(bytes);
|
||||
md5.write(bytes);
|
||||
}
|
||||
}
|
||||
Some(Err(err)) => {
|
||||
@@ -231,7 +230,7 @@ where
|
||||
|
||||
// println!("get len {}", bytes.len());
|
||||
// 如果有剩余
|
||||
if this.remaining.len() > 0 {
|
||||
if !this.remaining.is_empty() {
|
||||
let need_size = chuck_size - this.remaining.len();
|
||||
// 传入的数据大小需要补齐的大小,使用传入数据补齐
|
||||
if bytes.len() >= need_size {
|
||||
@@ -255,7 +254,7 @@ where
|
||||
this.streams.push_back(chuck);
|
||||
}
|
||||
|
||||
if bytes.len() > 0 {
|
||||
if !bytes.is_empty() {
|
||||
this.remaining.extend_from_slice(&bytes);
|
||||
}
|
||||
|
||||
@@ -267,21 +266,21 @@ where
|
||||
return Poll::Pending;
|
||||
}
|
||||
|
||||
if this.remaining.len() > 0 {
|
||||
if !this.remaining.is_empty() {
|
||||
let b = this.remaining.clone();
|
||||
this.remaining.clear();
|
||||
return Poll::Ready(Some(Ok(Bytes::from(b))));
|
||||
}
|
||||
Poll::Ready(None)
|
||||
}
|
||||
Err(err) => return Poll::Ready(Some(Err(err))),
|
||||
Err(err) => Poll::Ready(Some(Err(err))),
|
||||
},
|
||||
None => {
|
||||
// println!("get empty");
|
||||
if let Some(b) = this.streams.pop_front() {
|
||||
return Poll::Ready(Some(Ok(b)));
|
||||
}
|
||||
if this.remaining.len() > 0 {
|
||||
if !this.remaining.is_empty() {
|
||||
let b = this.remaining.clone();
|
||||
this.remaining.clear();
|
||||
return Poll::Ready(Some(Ok(Bytes::from(b))));
|
||||
@@ -343,7 +342,7 @@ where
|
||||
|
||||
fn size_hint(&self) -> (usize, Option<usize>) {
|
||||
let mut items = self.streams.len();
|
||||
if self.remaining.len() > 0 {
|
||||
if !self.remaining.is_empty() {
|
||||
items += 1;
|
||||
}
|
||||
(items, Some(items))
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
use std::str::FromStr;
|
||||
|
||||
use super::ecfs::FS;
|
||||
|
||||
use ecstore::bucket::policy::action::Action;
|
||||
@@ -42,7 +44,10 @@ impl S3Access for FS {
|
||||
// 上层验证了 ak/sk
|
||||
// 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 {
|
||||
card: cx.credentials().cloned(),
|
||||
|
||||
Reference in New Issue
Block a user