init NotificationSys

Cargo.lock

@@ -796,6 +796,7 @@ dependencies = [
  "http",
  "lazy_static",
  "lock",
+ "madmin",
  "md-5",
  "netif",
  "nix 0.29.0",
@@ -1672,7 +1673,6 @@ dependencies = [
 name = "madmin"
 version = "0.0.1"
 dependencies = [
- "ecstore",
  "psutil",
  "serde",
 ]

ecstore/Cargo.toml

@@ -63,6 +63,7 @@ rand.workspace = true
 pin-project-lite.workspace = true
 md-5.workspace = true
 workers.workspace = true
+madmin.workspace = true
 
 [target.'cfg(not(windows))'.dependencies]
 

ecstore/src/endpoints.rs

@@ -2,6 +2,7 @@ use crate::{
     disk::endpoint::{Endpoint, EndpointType},
     disks_layout::DisksLayout,
     error::{Error, Result},
+    global::global_rustfs_port,
     utils::net,
 };
 use std::{
@@ -529,6 +530,46 @@ impl EndpointServerPools {
 
         nodes
     }
+    pub fn hosts_sorted(&self) -> Vec<()> {
+        let (mut peers, local) = self.peers();
+
+        peers.sort();
+
+        for peer in peers.iter() {
+            if &local == peer {
+                continue;
+            }
+
+            // FIXME:TODO
+        }
+
+        unimplemented!()
+    }
+    pub fn peers(&self) -> (Vec<String>, String) {
+        let mut local = None;
+        let mut set = HashSet::new();
+        for ep in self.0.iter() {
+            for endpoint in ep.endpoints.0.iter() {
+                if endpoint.get_type() != EndpointType::Url {
+                    continue;
+                }
+                let host = endpoint.host_port();
+                if endpoint.is_local {
+                    if endpoint.url.port() == Some(global_rustfs_port()) {
+                        if local.is_none() {
+                            local = Some(host.clone());
+                        }
+                    }
+                }
+
+                set.insert(host);
+            }
+        }
+
+        let hosts: Vec<String> = set.iter().cloned().collect();
+
+        (hosts, local.unwrap_or_default())
+    }
 }
 
 #[cfg(test)]

ecstore/src/global.rs

@@ -18,7 +18,10 @@ pub const DISK_MIN_INODES: u64 = 1000;
 pub const DISK_FILL_FRACTION: f64 = 0.99;
 pub const DISK_RESERVE_FRACTION: f64 = 0.15;
 
+pub const DEFAULT_PORT: u16 = 9000;
+
 lazy_static! {
+    static ref GLOBAL_RUSTFS_PORT: OnceLock<u16> = OnceLock::new();
     pub static ref GLOBAL_OBJECT_API: OnceLock<Arc<ECStore>> = OnceLock::new();
     pub static ref GLOBAL_LOCAL_DISK: Arc<RwLock<Vec<Option<DiskStore>>>> = Arc::new(RwLock::new(Vec::new()));
     pub static ref GLOBAL_IsErasure: RwLock<bool> = RwLock::new(false);
@@ -34,6 +37,18 @@ lazy_static! {
     static ref globalDeploymentIDPtr: RwLock<Uuid> = RwLock::new(Uuid::nil());
 }
 
+pub fn global_rustfs_port() -> u16 {
+    if let Some(p) = GLOBAL_RUSTFS_PORT.get() {
+        p.clone()
+    } else {
+        DEFAULT_PORT
+    }
+}
+
+pub fn set_global_rustfs_port(value: u16) {
+    GLOBAL_RUSTFS_PORT.set(value).expect("set_global_rustfs_port fail");
+}
+
 pub async fn set_global_deployment_id(id: Uuid) {
     let mut id_ptr = globalDeploymentIDPtr.write().await;
     *id_ptr = id

ecstore/src/heal/heal_commands.rs

@@ -1,4 +1,8 @@
-use std::{path::Path, time::SystemTime};
+use std::{
+    collections::{HashMap, HashSet},
+    path::Path,
+    time::SystemTime,
+};
 
 use chrono::{DateTime, Utc};
 use lazy_static::lazy_static;
@@ -7,15 +11,18 @@ use time::OffsetDateTime;
 use tokio::sync::RwLock;
 
 use crate::{
+    config::storageclass::{RRS, STANDARD},
     disk::{DeleteOptions, DiskAPI, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
     error::{Error, Result},
     global::GLOBAL_BackgroundHealState,
     heal::heal_ops::HEALING_TRACKER_FILENAME,
     new_object_layer_fn,
-    store_api::{BucketInfo, StorageAPI},
+    store_api::{BucketInfo, StorageAPI, StorageDisk},
     utils::fs::read_file,
 };
 
+use super::{background_heal_ops::get_local_disks_to_heal, heal_ops::BG_HEALING_UUID};
+
 pub type HealScanMode = usize;
 pub type HealItemType = String;
 
@@ -494,3 +501,99 @@ pub async fn healing(derive_path: &str) -> Result<Option<HealingTracker>> {
 
     Ok(Some(healing_tracker))
 }
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+pub struct MRFStatus {
+    bytes_healed: u64,
+    items_healed: u64,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+pub struct SetStatus {
+    pub id: String,
+    pub pool_index: i32,
+    pub set_index: i32,
+    pub heal_status: String,
+    pub heal_priority: String,
+    pub total_objects: usize,
+    pub disks: Vec<StorageDisk>,
+}
+
+#[derive(Debug, Default, Serialize, Deserialize)]
+pub struct BgHealState {
+    offline_endpoints: Vec<String>,
+    scanned_items_count: u64,
+    heal_disks: Vec<String>,
+    sets: Vec<SetStatus>,
+    mrf: HashMap<String, MRFStatus>,
+    scparity: HashMap<String, usize>,
+}
+
+pub async fn get_local_background_heal_status() -> (BgHealState, bool) {
+    let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
+    if !ok {
+        return (BgHealState::default(), false);
+    }
+    let bg_seq = bg_seq.unwrap();
+    let mut status = BgHealState {
+        scanned_items_count: bg_seq.get_scanned_items_count().await as u64,
+        ..Default::default()
+    };
+    let mut heal_disks_map = HashSet::new();
+    for ep in get_local_disks_to_heal().await.iter() {
+        heal_disks_map.insert(ep.to_string());
+    }
+
+    let Some(store) = new_object_layer_fn() else {
+        let healing = GLOBAL_BackgroundHealState.get_local_healing_disks().await;
+        for disk in healing.values() {
+            status.heal_disks.push(disk.endpoint.clone());
+        }
+        return (status, true);
+    };
+
+    let si = store.local_storage_info().await;
+    let mut indexed = HashMap::new();
+    for disk in si.disks.iter() {
+        let set_idx = format!("{}-{}", disk.pool_index, disk.set_index);
+        // indexed.insert(set_idx, disk);
+        indexed.entry(set_idx).or_insert(Vec::new()).push(disk);
+    }
+
+    for (id, disks) in indexed {
+        let mut ss = SetStatus {
+            id,
+            set_index: disks[0].set_index,
+            pool_index: disks[0].pool_index,
+            ..Default::default()
+        };
+        for disk in disks {
+            ss.disks.push(disk.clone());
+            if disk.healing {
+                ss.heal_status = "healing".to_string();
+                ss.heal_priority = "high".to_string();
+                status.heal_disks.push(disk.endpoint.clone());
+            }
+        }
+        ss.disks.sort_by(|a, b| {
+            if a.pool_index != b.pool_index {
+                return a.pool_index.cmp(&b.pool_index);
+            }
+            if a.set_index != b.set_index {
+                return a.set_index.cmp(&b.set_index);
+            }
+            a.disk_index.cmp(&b.disk_index)
+        });
+        status.sets.push(ss);
+    }
+    status.sets.sort_by(|a, b| a.id.cmp(&b.id));
+    let backend_info = store.backend_info().await;
+    status
+        .scparity
+        .insert(STANDARD.to_string(), backend_info.standard_sc_parity.unwrap_or_default());
+    status
+        .scparity
+        .insert(RRS.to_string(), backend_info.rr_sc_parity.unwrap_or_default());
+
+    (status, true)
+}

ecstore/src/lib.rs

@@ -11,7 +11,9 @@ pub mod error;
 mod file_meta;
 pub mod global;
 pub mod heal;
+pub mod notification_sys;
 pub mod peer;
+mod peer_rest_client;
 mod quorum;
 pub mod set_disk;
 mod sets;
@@ -22,7 +24,7 @@ pub mod utils;
 
 pub mod bucket;
 pub mod file_meta_inline;
-pub mod options;
+
 pub mod pools;
 pub mod store_err;
 pub mod xhttp;

ecstore/src/notification_sys.rs

@@ -0,0 +1,53 @@
+use std::sync::OnceLock;
+
+use crate::endpoints::EndpointServerPools;
+use crate::error::{Error, Result};
+use crate::peer_rest_client::PeerRestClient;
+use lazy_static::lazy_static;
+
+lazy_static! {
+    pub static ref GLOBAL_NotificationSys: OnceLock<NotificationSys> = OnceLock::new();
+}
+
+pub async fn new_global_notification_sys(eps: EndpointServerPools) -> Result<()> {
+    let _ = GLOBAL_NotificationSys
+        .set(NotificationSys::new(eps).await)
+        .map_err(|_| Error::msg("init notification_sys fail"));
+    Ok(())
+}
+
+pub struct NotificationSys {
+    pub peer_clients: Vec<PeerRestClient>,
+    pub all_peer_clients: Vec<PeerRestClient>,
+}
+
+impl NotificationSys {
+    pub async fn new(eps: EndpointServerPools) -> Self {
+        let (peer_clients, all_peer_clients) = PeerRestClient::new_clients(eps).await;
+        Self {
+            peer_clients,
+            all_peer_clients,
+        }
+    }
+}
+
+pub struct NotificationPeerErr {
+    pub host: String,
+    pub err: Option<Error>,
+}
+
+impl NotificationSys {
+    pub async fn delete_policy(&self) -> Vec<NotificationPeerErr> {
+        unimplemented!()
+    }
+    pub async fn load_policy(&self) -> Vec<NotificationPeerErr> {
+        unimplemented!()
+    }
+
+    pub async fn load_policy_mapping(&self) -> Vec<NotificationPeerErr> {
+        unimplemented!()
+    }
+    pub async fn delete_user(&self) -> Vec<NotificationPeerErr> {
+        unimplemented!()
+    }
+}

ecstore/src/peer_rest_client.rs

@@ -1,9 +1,11 @@
 use std::{collections::HashMap, io::Cursor, time::SystemTime};
 
+use crate::{
+    admin_server_info::ServerProperties, endpoints::EndpointServerPools, global::is_dist_erasure,
+    heal::heal_commands::BgHealState, store_api::StorageInfo,
+};
 use common::error::{Error, Result};
-use ecstore::{admin_server_info::ServerProperties, store_api::StorageInfo};
 use madmin::{
-    heal_command::BgHealState,
     health::{Cpus, MemInfo, OsInfo, Partitions, ProcInfo, SysConfig, SysErrors, SysService},
     metrics::{CollectMetricsOpts, MetricType, RealtimeMetrics},
     net::NetInfo,
@@ -21,14 +23,14 @@ use protos::{
     },
 };
 use rmp_serde::{Deserializer, Serializer};
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize as _};
 use tonic::Request;
 
 pub const PEER_RESTSIGNAL: &str = "signal";
 pub const PEER_RESTSUB_SYS: &str = "sub-sys";
 pub const PEER_RESTDRY_RUN: &str = "dry-run";
 
-struct PeerRestClient {
+pub struct PeerRestClient {
     addr: String,
 }
 
@@ -38,6 +40,15 @@ impl PeerRestClient {
             addr: format!("{}://{}:{}", url.scheme(), url.host_str().unwrap(), url.port().unwrap()),
         }
     }
+    pub async fn new_clients(_eps: EndpointServerPools) -> (Vec<Self>, Vec<Self>) {
+        if !is_dist_erasure().await {
+            return (Vec::new(), Vec::new());
+        }
+
+        // FIXME:TODO
+
+        todo!()
+    }
 }
 
 impl PeerRestClient {

ecstore/src/utils/net.rs

@@ -2,8 +2,9 @@ use crate::error::{Error, Result};
 use lazy_static::lazy_static;
 use std::{
     collections::HashSet,
-    net::{IpAddr, SocketAddr, ToSocketAddrs},
+    net::{IpAddr, SocketAddr, TcpListener, ToSocketAddrs},
 };
+
 use url::Host;
 
 lazy_static! {
@@ -92,6 +93,10 @@ pub fn get_host_ip(host: Host<&str>) -> Result<HashSet<IpAddr>> {
     }
 }
 
+pub fn get_available_port() -> u16 {
+    TcpListener::bind("0.0.0.0:0").unwrap().local_addr().unwrap().port()
+}
+
 /// returns IPs of local interface
 pub(crate) fn must_get_local_ips() -> Result<Vec<IpAddr>> {
     match netif::up() {

madmin/Cargo.toml

@@ -7,6 +7,5 @@ rust-version.workspace = true
 version.workspace = true
 
 [dependencies]
-ecstore.workspace = true
 psutil = "3.3.0"
 serde.workspace = true

madmin/src/heal_command.rs

@@ -1,106 +0,0 @@
-use std::collections::{HashMap, HashSet};
-
-use ecstore::{
-    config::storageclass::{RRS, STANDARD},
-    global::GLOBAL_BackgroundHealState,
-    heal::{background_heal_ops::get_local_disks_to_heal, heal_ops::BG_HEALING_UUID},
-    new_object_layer_fn,
-    store_api::{StorageAPI, StorageDisk},
-};
-use serde::{Deserialize, Serialize};
-
-#[derive(Debug, Default, Serialize, Deserialize)]
-pub struct MRFStatus {
-    bytes_healed: u64,
-    items_healed: u64,
-}
-
-#[derive(Debug, Default, Serialize, Deserialize)]
-pub struct SetStatus {
-    pub id: String,
-    pub pool_index: i32,
-    pub set_index: i32,
-    pub heal_status: String,
-    pub heal_priority: String,
-    pub total_objects: usize,
-    pub disks: Vec<StorageDisk>,
-}
-
-#[derive(Debug, Default, Serialize, Deserialize)]
-pub struct BgHealState {
-    offline_endpoints: Vec<String>,
-    scanned_items_count: u64,
-    heal_disks: Vec<String>,
-    sets: Vec<SetStatus>,
-    mrf: HashMap<String, MRFStatus>,
-    scparity: HashMap<String, usize>,
-}
-
-pub async fn get_local_background_heal_status() -> (BgHealState, bool) {
-    let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
-    if !ok {
-        return (BgHealState::default(), false);
-    }
-    let bg_seq = bg_seq.unwrap();
-    let mut status = BgHealState {
-        scanned_items_count: bg_seq.get_scanned_items_count().await as u64,
-        ..Default::default()
-    };
-    let mut heal_disks_map = HashSet::new();
-    for ep in get_local_disks_to_heal().await.iter() {
-        heal_disks_map.insert(ep.to_string());
-    }
-
-    let Some(store) = new_object_layer_fn() else {
-        let healing = GLOBAL_BackgroundHealState.get_local_healing_disks().await;
-        for disk in healing.values() {
-            status.heal_disks.push(disk.endpoint.clone());
-        }
-        return (status, true);
-    };
-
-    let si = store.local_storage_info().await;
-    let mut indexed = HashMap::new();
-    for disk in si.disks.iter() {
-        let set_idx = format!("{}-{}", disk.pool_index, disk.set_index);
-        // indexed.insert(set_idx, disk);
-        indexed.entry(set_idx).or_insert(Vec::new()).push(disk);
-    }
-
-    for (id, disks) in indexed {
-        let mut ss = SetStatus {
-            id,
-            set_index: disks[0].set_index,
-            pool_index: disks[0].pool_index,
-            ..Default::default()
-        };
-        for disk in disks {
-            ss.disks.push(disk.clone());
-            if disk.healing {
-                ss.heal_status = "healing".to_string();
-                ss.heal_priority = "high".to_string();
-                status.heal_disks.push(disk.endpoint.clone());
-            }
-        }
-        ss.disks.sort_by(|a, b| {
-            if a.pool_index != b.pool_index {
-                return a.pool_index.cmp(&b.pool_index);
-            }
-            if a.set_index != b.set_index {
-                return a.set_index.cmp(&b.set_index);
-            }
-            a.disk_index.cmp(&b.disk_index)
-        });
-        status.sets.push(ss);
-    }
-    status.sets.sort_by(|a, b| a.id.cmp(&b.id));
-    let backend_info = store.backend_info().await;
-    status
-        .scparity
-        .insert(STANDARD.to_string(), backend_info.standard_sc_parity.unwrap_or_default());
-    status
-        .scparity
-        .insert(RRS.to_string(), backend_info.rr_sc_parity.unwrap_or_default());
-
-    (status, true)
-}

madmin/src/lib.rs

@@ -1,4 +1,3 @@
-pub mod heal_command;
 pub mod health;
 pub mod metrics;
 pub mod net;

rustfs/src/config/mod.rs

@@ -1,12 +1,13 @@
 use clap::Parser;
 use const_str::concat;
+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_PORT: u16 = 9000;
+
 pub const DEFAULT_ACCESS_KEY: &str = "rustfsadmin";
 pub const DEFAULT_SECRET_KEY: &str = "rustfsadmin";
 

rustfs/src/grpc.rs

@@ -13,7 +13,10 @@ use ecstore::{
         UpdateMetadataOpts, WalkDirOptions,
     },
     erasure::Writer,
-    heal::{data_usage_cache::DataUsageCache, heal_commands::HealOpts},
+    heal::{
+        data_usage_cache::DataUsageCache,
+        heal_commands::{get_local_background_heal_status, HealOpts},
+    },
     new_object_layer_fn,
     peer::{LocalPeerS3Client, PeerS3Client},
     store::{all_local_disk_path, find_local_disk},
@@ -25,7 +28,6 @@ use lock::{lock_args::LockArgs, Locker, GLOBAL_LOCAL_SERVER};
 use common::globals::GLOBAL_Local_Node_Name;
 use madmin::net::get_net_info;
 use madmin::{
-    heal_command::get_local_background_heal_status,
     health::{
         get_cpus, get_mem_info, get_os_info, get_partitions, get_proc_info, get_sys_config, get_sys_errors, get_sys_services,
     },

rustfs/src/main.rs

@@ -1,16 +1,17 @@
 mod admin;
 mod config;
 mod grpc;
-#[allow(dead_code)]
-mod peer_rest_client;
 mod service;
 mod storage;
-
 use clap::Parser;
 use common::error::{Error, Result};
+use ecstore::global::set_global_rustfs_port;
+use ecstore::heal::background_heal_ops::init_auto_heal;
+use ecstore::utils::net::{self, get_available_port};
 use ecstore::{
     endpoints::EndpointServerPools,
-    heal::{background_heal_ops::init_auto_heal, data_scanner::init_data_scanner},
+    heal::data_scanner::init_data_scanner,
+    notification_sys::new_global_notification_sys,
     set_global_endpoints,
     store::{init_local_disks, ECStore},
     update_erasure_type,
@@ -72,13 +73,27 @@ fn main() -> Result<()> {
 async fn run(opt: config::Opt) -> Result<()> {
     debug!("opt: {:?}", &opt);
 
+    let mut server_addr = net::check_local_server_addr(opt.address.as_str()).unwrap();
+
+    if server_addr.port() == 0 {
+        server_addr.set_port(get_available_port());
+    }
+
+    let server_port = server_addr.port();
+
+    let server_address = server_addr.to_string();
+
+    debug!("server_address {}", &server_address);
+
+    set_global_rustfs_port(server_port);
+
     //监听地址,端口从参数中获取
-    let listener = TcpListener::bind(opt.address.clone()).await?;
+    let listener = TcpListener::bind(server_address.clone()).await?;
     //获取监听地址
     let local_addr: SocketAddr = listener.local_addr()?;
 
     // 用于rpc
-    let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(opt.address.clone().as_str(), opt.volumes.clone())
+    let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address.clone().as_str(), opt.volumes.clone())
         .map_err(|err| Error::from_string(err.to_string()))?;
 
     for (i, eps) in endpoint_pools.as_ref().iter().enumerate() {
@@ -100,7 +115,7 @@ async fn run(opt: config::Opt) -> Result<()> {
     // 本项目使用s3s库来实现s3服务
     let service = {
         let store = storage::ecfs::FS::new();
-        // let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(opt.address.clone(), endpoint_pools).await?);
+        // let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(server_address.clone(), endpoint_pools).await?);
         let mut b = S3ServiceBuilder::new(store.clone());
         //设置AK和SK
         //其中部份内容从config配置文件中读取
@@ -183,7 +198,7 @@ async fn run(opt: config::Opt) -> Result<()> {
     });
 
     // init store
-    let store = ECStore::new(opt.address.clone(), endpoint_pools.clone())
+    let store = ECStore::new(server_address.clone(), endpoint_pools.clone())
         .await
         .map_err(|err| Error::from_string(err.to_string()))?;
 
@@ -192,6 +207,12 @@ async fn run(opt: config::Opt) -> Result<()> {
         Error::from_string(err.to_string())
     })?;
     warn!(" init store success!");
+
+    new_global_notification_sys(endpoint_pools.clone()).await.map_err(|err| {
+        error!("new_global_notification_sys faild {:?}", &err);
+        Error::from_string(err.to_string())
+    })?;
+
     // init scanner
     init_data_scanner().await;
     // init auto heal

rustfs/src/storage/ecfs.rs

@@ -1,3 +1,6 @@
+use super::options::del_opts;
+use super::options::extract_metadata;
+use super::options::put_opts;
 use bytes::Bytes;
 use common::error::Result;
 use ecstore::bucket::error::BucketMetadataError;
@@ -16,9 +19,6 @@ use ecstore::bucket::tagging::decode_tags;
 use ecstore::bucket::tagging::encode_tags;
 use ecstore::bucket::versioning_sys::BucketVersioningSys;
 use ecstore::new_object_layer_fn;
-use ecstore::options::del_opts;
-use ecstore::options::extract_metadata;
-use ecstore::options::put_opts;
 use ecstore::store_api::BucketOptions;
 use ecstore::store_api::CompletePart;
 use ecstore::store_api::DeleteBucketOptions;

rustfs/src/storage/mod.rs

@@ -1,3 +1,4 @@
 pub mod acess;
 pub mod ecfs;
 pub mod error;
+pub mod options;

rustfs/src/storage/options.rs

@@ -1,8 +1,8 @@
-use crate::bucket::versioning_sys::BucketVersioningSys;
-use crate::error::{Error, Result};
-use crate::store_api::ObjectOptions;
-use crate::store_err::StorageError;
-use crate::utils::path::is_dir_object;
+use ecstore::bucket::versioning_sys::BucketVersioningSys;
+use ecstore::error::{Error, Result};
+use ecstore::store_api::ObjectOptions;
+use ecstore::store_err::StorageError;
+use ecstore::utils::path::is_dir_object;
 use http::{HeaderMap, HeaderValue};
 use lazy_static::lazy_static;
 use std::collections::HashMap;