test StorageInfoHandler

ecstore/src/notification_sys.rs

@@ -1,8 +1,11 @@
-use std::sync::OnceLock;
+use std::sync::{Arc, OnceLock};
 
 use crate::endpoints::EndpointServerPools;
 use crate::error::{Error, Result};
+use crate::global::get_global_endpoints;
 use crate::peer_rest_client::PeerRestClient;
+use crate::StorageAPI;
+use futures::future::join_all;
 use lazy_static::lazy_static;
 
 lazy_static! {
@@ -16,9 +19,13 @@ pub async fn new_global_notification_sys(eps: EndpointServerPools) -> Result<()>
     Ok(())
 }
 
+pub fn get_global_notification_sys() -> Option<&'static NotificationSys> {
+    GLOBAL_NotificationSys.get()
+}
+
 pub struct NotificationSys {
-    pub peer_clients: Vec<PeerRestClient>,
-    pub all_peer_clients: Vec<PeerRestClient>,
+    peer_clients: Vec<Option<PeerRestClient>>,
+    all_peer_clients: Vec<Option<PeerRestClient>>,
 }
 
 impl NotificationSys {
@@ -50,4 +57,57 @@ impl NotificationSys {
     pub async fn delete_user(&self) -> Vec<NotificationPeerErr> {
         unimplemented!()
     }
+
+    pub async fn storage_info<S: StorageAPI>(&self, api: &S) -> madmin::StorageInfo {
+        let mut futures = Vec::with_capacity(self.peer_clients.len());
+
+        for client in self.peer_clients.iter() {
+            futures.push(async move {
+                if let Some(client) = client {
+                    match client.local_storage_info().await {
+                        Ok(info) => Some(info),
+                        Err(_) => Some(madmin::StorageInfo {
+                            disks: get_offline_disks(&client.host.to_string(), &get_global_endpoints()),
+                            ..Default::default()
+                        }),
+                    }
+                } else {
+                    None
+                }
+            });
+        }
+
+        let mut replies = join_all(futures).await;
+
+        replies.push(Some(api.local_storage_info().await));
+
+        let mut disks = Vec::new();
+        for info in replies.into_iter().flatten() {
+            disks.extend(info.disks);
+        }
+
+        let backend = api.backend_info().await;
+        madmin::StorageInfo { disks, backend }
+    }
+}
+
+fn get_offline_disks(offline_host: &str, endpoints: &EndpointServerPools) -> Vec<madmin::Disk> {
+    let mut offline_disks = Vec::new();
+
+    for pool in endpoints.as_ref() {
+        for ep in pool.endpoints.as_ref() {
+            if (offline_host.is_empty() && ep.is_local) || offline_host == ep.host_port() {
+                offline_disks.push(madmin::Disk {
+                    endpoint: ep.to_string(),
+                    state: madmin::ItemState::Offline.to_string().to_owned(),
+                    pool_index: ep.pool_idx,
+                    set_index: ep.set_idx,
+                    disk_index: ep.disk_idx,
+                    ..Default::default()
+                });
+            }
+        }
+    }
+
+    offline_disks
 }