notification_sys base

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!();
         }
     }
 }

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;
 

ecstore/src/disk/error.rs

@@ -139,13 +139,12 @@ impl DiskError {
     }
 
     pub fn count_errs(&self, errs: &[Option<Error>]) -> 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<Error>]) -> bool {

ecstore/src/disk/mod.rs

@@ -994,6 +994,7 @@ impl BufferWriter {
     pub fn new(inner: Vec<u8>) -> 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<Channel, Box<dyn Fn(Request<()>) -> Result<Request<()>, 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<Channel, Box<dyn Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static>>,
-        >,
-    ) -> Result<Self> {
+    pub async fn new(root: PathBuf, volume: String, path: String, is_append: bool, mut client: NodeClient) -> Result<Self> {
         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<Channel, Box<dyn Fn(Request<()>) -> Result<Request<()>, Status> + Send + Sync + 'static>>,
-        >,
-    ) -> Result<Self> {
+    pub async fn new(root: PathBuf, volume: String, path: String, mut client: NodeClient) -> Result<Self> {
         let (tx, rx) = mpsc::channel(128);
         let in_stream = ReceiverStream::new(rx);
 

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需要补全，所以最后一个长度应该也是一样的

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;

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<Arc<ECStore>> {
-    GLOBAL_OBJECT_API.get().map(|ec| ec.clone())
+    GLOBAL_OBJECT_API.get().cloned()
 }
 
 pub async fn set_object_layer(o: Arc<ECStore>) {

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);

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<NotificationSys> = 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<madmin::Disk> {

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 {

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(())
     }

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<ObjectToDelete>,
         opts: ObjectOptions,
     ) -> Result<(Vec<DeletedObject>, Vec<Option<Error>>)>;
-    #[warn(clippy::too_many_arguments)]
+
     // TransitionObject TODO:
     // RestoreTransitionedObject TODO:
 

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));
 }

ecstore/src/utils/os/unix.rs

@@ -77,6 +77,6 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result<bool> {
     Ok(stat1.st_dev == stat2.st_dev)
 }
 
-pub fn get_drive_stats(major: u32, minor: u32) -> Result<IOStats> {
+pub fn get_drive_stats(_major: u32, _minor: u32) -> Result<IOStats> {
     Ok(IOStats::default())
 }

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();
 

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";
 

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