diff --git a/common/workers/src/workers.rs b/common/workers/src/workers.rs index 5c03447f8..85be5efbc 100644 --- a/common/workers/src/workers.rs +++ b/common/workers/src/workers.rs @@ -81,7 +81,7 @@ mod tests { sleep(Duration::from_secs(1)).await; workers.wait().await; if workers.available().await != workers.limit { - assert!(false); + unreachable!(); } } } diff --git a/ecstore/src/bucket/policy/resource.rs b/ecstore/src/bucket/policy/resource.rs index 409cd9cb7..b9c520f5e 100644 --- a/ecstore/src/bucket/policy/resource.rs +++ b/ecstore/src/bucket/policy/resource.rs @@ -169,6 +169,7 @@ impl<'de> Deserialize<'de> for Resource { { struct Visitor; + #[allow(clippy::needless_lifetimes)] impl<'de> serde::de::Visitor<'de> for Visitor { type Value = Resource; diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index 0902e5d05..36e49f46c 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -139,13 +139,12 @@ impl DiskError { } pub fn count_errs(&self, errs: &[Option]) -> usize { - return errs - .iter() + errs.iter() .filter(|&err| match err { None => false, Some(e) => self.is(e), }) - .count(); + .count() } pub fn quorum_unformatted_disks(errs: &[Option]) -> bool { diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index ab57ce45a..dd1fc6c09 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -994,6 +994,7 @@ impl BufferWriter { pub fn new(inner: Vec) -> Self { Self { inner } } + #[allow(clippy::should_implement_trait)] pub fn as_ref(&self) -> &[u8] { self.inner.as_ref() } @@ -1038,6 +1039,10 @@ impl Writer for LocalFileWriter { } } +type NodeClient = NodeServiceClient< + InterceptedService) -> Result, Status> + Send + Sync + 'static>>, +>; + #[derive(Debug)] pub struct RemoteFileWriter { pub root: PathBuf, @@ -1049,15 +1054,7 @@ pub struct RemoteFileWriter { } impl RemoteFileWriter { - pub async fn new( - root: PathBuf, - volume: String, - path: String, - is_append: bool, - mut client: NodeServiceClient< - InterceptedService) -> Result, Status> + Send + Sync + 'static>>, - >, - ) -> Result { + pub async fn new(root: PathBuf, volume: String, path: String, is_append: bool, mut client: NodeClient) -> Result { let (tx, rx) = mpsc::channel(128); let in_stream = ReceiverStream::new(rx); @@ -1247,14 +1244,7 @@ pub struct RemoteFileReader { } impl RemoteFileReader { - pub async fn new( - root: PathBuf, - volume: String, - path: String, - mut client: NodeServiceClient< - InterceptedService) -> Result, Status> + Send + Sync + 'static>>, - >, - ) -> Result { + pub async fn new(root: PathBuf, volume: String, path: String, mut client: NodeClient) -> Result { let (tx, rx) = mpsc::channel(128); let in_stream = ReceiverStream::new(rx); diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index 9691a763a..bd559e499 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -387,7 +387,7 @@ impl Erasure { // 算出每个分片大小 pub fn shard_size(&self, data_size: usize) -> usize { - (data_size + self.data_shards - 1) / self.data_shards + data_size.div_ceil(self.data_shards) } // returns final erasure size from original size. pub fn shard_file_size(&self, total_size: usize) -> usize { @@ -397,7 +397,7 @@ impl Erasure { let num_shards = total_size / self.block_size; let last_block_size = total_size % self.block_size; - let last_shard_size = (last_block_size + self.data_shards - 1) / self.data_shards; + let last_shard_size = last_block_size.div_ceil(self.data_shards); num_shards * self.shard_size(self.block_size) + last_shard_size // // 因为写入的时候ec需要补全,所以最后一个长度应该也是一样的 diff --git a/ecstore/src/file_meta.rs b/ecstore/src/file_meta.rs index d506e3efa..3afae0557 100644 --- a/ecstore/src/file_meta.rs +++ b/ecstore/src/file_meta.rs @@ -250,7 +250,7 @@ impl FileMeta { // TODO: use old buf let meta_buf = ver.marshal_msg()?; - let pre_mod_time = self.versions[idx].header.mod_time.clone(); + let pre_mod_time = self.versions[idx].header.mod_time; self.versions[idx].header = ver.header(); self.versions[idx].meta = meta_buf; diff --git a/ecstore/src/global.rs b/ecstore/src/global.rs index 81f1a0d3b..f5a817c4c 100644 --- a/ecstore/src/global.rs +++ b/ecstore/src/global.rs @@ -39,7 +39,7 @@ lazy_static! { pub fn global_rustfs_port() -> u16 { if let Some(p) = GLOBAL_RUSTFS_PORT.get() { - p.clone() + *p } else { DEFAULT_PORT } @@ -74,7 +74,7 @@ pub fn get_global_endpoints() -> EndpointServerPools { } pub fn new_object_layer_fn() -> Option> { - GLOBAL_OBJECT_API.get().map(|ec| ec.clone()) + GLOBAL_OBJECT_API.get().cloned() } pub async fn set_object_layer(o: Arc) { diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 721a09f84..f3d1745d9 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -117,10 +117,7 @@ async fn run_data_scanner() { .map_or(Vec::new(), |buf| buf); let mut buf_t = Deserializer::new(Cursor::new(buf)); - let mut cycle_info: CurrentScannerCycle = match Deserialize::deserialize(&mut buf_t) { - Ok(info) => info, - Err(_) => CurrentScannerCycle::default(), - }; + let mut cycle_info: CurrentScannerCycle = Deserialize::deserialize(&mut buf_t).unwrap_or_default(); loop { let stop_fn = ScannerMetrics::log(ScannerMetric::ScanCycle); diff --git a/ecstore/src/notification_sys.rs b/ecstore/src/notification_sys.rs index 359e4c3f0..a36d14629 100644 --- a/ecstore/src/notification_sys.rs +++ b/ecstore/src/notification_sys.rs @@ -1,4 +1,4 @@ -use std::sync::{Arc, OnceLock}; +use std::sync::OnceLock; use crate::endpoints::EndpointServerPools; use crate::error::{Error, Result}; @@ -7,6 +7,7 @@ use crate::peer_rest_client::PeerRestClient; use crate::StorageAPI; use futures::future::join_all; use lazy_static::lazy_static; +use tracing::error; lazy_static! { pub static ref GLOBAL_NotificationSys: OnceLock = OnceLock::new(); @@ -89,6 +90,20 @@ impl NotificationSys { let backend = api.backend_info().await; madmin::StorageInfo { disks, backend } } + + pub async fn reload_pool_meta(&self) { + let mut futures = Vec::with_capacity(self.peer_clients.len()); + for client in self.peer_clients.iter().flatten() { + futures.push(client.reload_pool_meta()); + } + + let results = join_all(futures).await; + for result in results { + if let Err(err) = result { + error!("notification reload_pool_meta err {:?}", err); + } + } + } } fn get_offline_disks(offline_host: &str, endpoints: &EndpointServerPools) -> Vec { diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index 61737fc04..a5217ba95 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -110,7 +110,7 @@ impl S3PeerSys { let mut futures = Vec::new(); let heal_bucket_results = Arc::new(RwLock::new(vec![HealResultItem::default(); self.clients.len()])); for (idx, client) in self.clients.iter().enumerate() { - let opts_clone = opts.clone(); + let opts_clone = opts; let heal_bucket_results_clone = heal_bucket_results.clone(); futures.push(async move { match client.heal_bucket(bucket, &opts_clone).await { diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index d355141a7..d6a43f83a 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -5,6 +5,7 @@ use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::heal::heal_commands::HealOpts; use crate::new_object_layer_fn; +use crate::notification_sys::get_global_notification_sys; use crate::store_api::{BucketOptions, MakeBucketOptions, StorageAPI}; use crate::store_err::{is_err_bucket_exists, StorageError}; use crate::utils::path::{path_join, SLASH_SEPARATOR}; @@ -12,6 +13,7 @@ use crate::{sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use rmp_serde::{Deserializer, Serializer}; use serde::{Deserialize, Serialize}; +use std::fmt::Display; use std::io::{Cursor, Write}; use std::path::PathBuf; use std::sync::Arc; @@ -270,7 +272,7 @@ impl PoolMeta { } } -fn path2_bucket_object(name: &String) -> (String, String) { +fn path2_bucket_object(name: &str) -> (String, String) { path2_bucket_object_with_base_path("", name) } @@ -374,11 +376,13 @@ pub struct DecomBucketInfo { pub prefix: String, } -impl ToString for DecomBucketInfo { - fn to_string(&self) -> String { - path_join(&[PathBuf::from(self.name.clone()), PathBuf::from(self.prefix.clone())]) - .to_string_lossy() - .to_string() +impl Display for DecomBucketInfo { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!( + f, + "{}", + path_join(&[PathBuf::from(self.name.clone()), PathBuf::from(self.prefix.clone())]).to_string_lossy() + ) } } @@ -531,7 +535,9 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_failed(idx) { pool_meta.save(self.pools.clone()).await?; - // FIXME: globalNotificationSys.ReloadPoolMeta(ctx) + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } } Ok(()) @@ -545,7 +551,9 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_complete(idx) { pool_meta.save(self.pools.clone()).await?; - // FIXME: globalNotificationSys.ReloadPoolMeta(ctx) + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } } Ok(()) @@ -641,7 +649,9 @@ impl ECStore { pool_meta.save(self.pools.clone()).await?; - // FIXME: globalNotificationSys.ReloadPoolMeta(ctx) + if let Some(notification_sys) = get_global_notification_sys() { + notification_sys.reload_pool_meta().await; + } Ok(()) } diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 8b34fde3c..715d21810 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -331,7 +331,7 @@ impl ErasureInfo { // 算出每个分片大小 pub fn shard_size(&self, data_size: usize) -> usize { - (data_size + self.data_blocks - 1) / self.data_blocks + data_size.div_ceil(self.data_blocks) } // returns final erasure size from original size. @@ -342,7 +342,7 @@ impl ErasureInfo { let num_shards = total_size / self.block_size; let last_block_size = total_size % self.block_size; - let last_shard_size = (last_block_size + self.data_blocks - 1) / self.data_blocks; + let last_shard_size = last_block_size.div_ceil(self.data_blocks); num_shards * self.shard_size(self.block_size) + last_shard_size // // 因为写入的时候ec需要补全,所以最后一个长度应该也是一样的 @@ -826,6 +826,7 @@ pub trait ObjectIO: Send + Sync + 'static { } #[async_trait::async_trait] +#[allow(clippy::too_many_arguments)] pub trait StorageAPI: ObjectIO { // NewNSLock TODO: // Shutdown TODO: @@ -873,7 +874,7 @@ pub trait StorageAPI: ObjectIO { objects: Vec, opts: ObjectOptions, ) -> Result<(Vec, Vec>)>; - #[warn(clippy::too_many_arguments)] + // TransitionObject TODO: // RestoreTransitionedObject TODO: diff --git a/ecstore/src/store_err.rs b/ecstore/src/store_err.rs index 5ee591dd5..e0dabb870 100644 --- a/ecstore/src/store_err.rs +++ b/ecstore/src/store_err.rs @@ -208,7 +208,7 @@ pub fn is_err_object_not_found(err: &Error) -> bool { 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); + assert!(is_err_bucket_exists(&e1)); + assert!(!is_err_object_not_found(&e1)); + assert!(is_err_object_not_found(&e2)); } diff --git a/ecstore/src/utils/os/unix.rs b/ecstore/src/utils/os/unix.rs index 17ccd06f1..bee857150 100644 --- a/ecstore/src/utils/os/unix.rs +++ b/ecstore/src/utils/os/unix.rs @@ -77,6 +77,6 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result { Ok(stat1.st_dev == stat2.st_dev) } -pub fn get_drive_stats(major: u32, minor: u32) -> Result { +pub fn get_drive_stats(_major: u32, _minor: u32) -> Result { Ok(IOStats::default()) } diff --git a/madmin/src/metrics.rs b/madmin/src/metrics.rs index fee5b365c..f8c6948c9 100644 --- a/madmin/src/metrics.rs +++ b/madmin/src/metrics.rs @@ -168,7 +168,7 @@ impl ScannerMetrics { self.last_minute.ilm.entry(k.clone()).or_default().merge(v); } - self.active_paths.extend(other.active_paths.clone().into_iter()); + self.active_paths.extend(other.active_paths.clone()); self.active_paths.sort(); } @@ -606,14 +606,14 @@ pub struct RealtimeMetrics { impl RealtimeMetrics { pub fn merge(&mut self, other: Self) { if !other.errors.is_empty() { - self.errors.extend(other.errors.into_iter()); + self.errors.extend(other.errors); } for (k, v) in other.by_host.into_iter() { *self.by_host.entry(k).or_default() = v; } - self.hosts.extend(other.hosts.into_iter()); + self.hosts.extend(other.hosts); self.aggregated.merge(&other.aggregated); self.hosts.sort(); diff --git a/rustfs/src/config/mod.rs b/rustfs/src/config/mod.rs index f038b3b57..c87957cdb 100644 --- a/rustfs/src/config/mod.rs +++ b/rustfs/src/config/mod.rs @@ -4,10 +4,6 @@ use ecstore::global::DEFAULT_PORT; shadow_rs::shadow!(build); -/// Default port that a rustfs server listens on. -/// -/// Used if no port is specified. - pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin"; pub const DEFAULT_SECRET_KEY: &str = "rustfsadmin"; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index cd1da10e6..b3a40ac87 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -591,8 +591,6 @@ impl S3 for FS { .await .map_err(to_s3_error)?; - error!("ObjectOptions {:?}", opts); - let obj_info = store .put_object(&bucket, &key, &mut reader, &opts) .await