refactor: reorganize RPC modules into unified structure

ecstore/src/rpc/http_auth.rs

@@ -0,0 +1,375 @@
+use crate::global::get_global_action_cred;
+use base64::Engine as _;
+use base64::engine::general_purpose;
+use hmac::{Hmac, Mac};
+use http::HeaderMap;
+use http::HeaderValue;
+use http::Method;
+use http::Uri;
+use sha2::Sha256;
+use time::OffsetDateTime;
+use tracing::error;
+
+type HmacSha256 = Hmac<Sha256>;
+
+const SIGNATURE_HEADER: &str = "x-rustfs-signature";
+const TIMESTAMP_HEADER: &str = "x-rustfs-timestamp";
+const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes
+
+/// Get the shared secret for HMAC signing
+fn get_shared_secret() -> String {
+    if let Some(cred) = get_global_action_cred() {
+        cred.secret_key
+    } else {
+        // Fallback to environment variable if global credentials are not available
+        std::env::var("RUSTFS_RPC_SECRET").unwrap_or_else(|_| "rustfs-default-secret".to_string())
+    }
+}
+
+/// Generate HMAC-SHA256 signature for the given data
+fn generate_signature(secret: &str, url: &str, method: &Method, timestamp: i64) -> String {
+    let uri: Uri = url.parse().expect("Invalid URL");
+
+    let path_and_query = uri.path_and_query().unwrap();
+
+    let url = path_and_query.to_string();
+
+    let data = format!("{}|{}|{}", url, method, timestamp);
+    let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC can take key of any size");
+    mac.update(data.as_bytes());
+    let result = mac.finalize();
+    general_purpose::STANDARD.encode(result.into_bytes())
+}
+
+/// Build headers with authentication signature
+pub fn build_auth_headers(url: &str, method: &Method, headers: &mut HeaderMap) {
+    let secret = get_shared_secret();
+    let timestamp = OffsetDateTime::now_utc().unix_timestamp();
+
+    let signature = generate_signature(&secret, url, method, timestamp);
+
+    headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+    headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&timestamp.to_string()).unwrap());
+}
+
+/// Verify the request signature for RPC requests
+pub fn verify_rpc_signature(url: &str, method: &Method, headers: &HeaderMap) -> std::io::Result<()> {
+    let secret = get_shared_secret();
+
+    // Get signature from header
+    let signature = headers
+        .get(SIGNATURE_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .ok_or_else(|| std::io::Error::other("Missing signature header"))?;
+
+    // Get timestamp from header
+    let timestamp_str = headers
+        .get(TIMESTAMP_HEADER)
+        .and_then(|v| v.to_str().ok())
+        .ok_or_else(|| std::io::Error::other("Missing timestamp header"))?;
+
+    let timestamp: i64 = timestamp_str
+        .parse()
+        .map_err(|_| std::io::Error::other("Invalid timestamp format"))?;
+
+    // Check timestamp validity (prevent replay attacks)
+    let current_time = OffsetDateTime::now_utc().unix_timestamp();
+
+    if current_time.saturating_sub(timestamp) > SIGNATURE_VALID_DURATION {
+        return Err(std::io::Error::other("Request timestamp expired"));
+    }
+
+    // Generate expected signature
+
+    let expected_signature = generate_signature(&secret, url, method, timestamp);
+
+    // Compare signatures
+    if signature != expected_signature {
+        error!(
+            "verify_rpc_signature: Invalid signature: secret {}, url {}, method {}, timestamp {}, signature {}, expected_signature {}",
+            secret, url, method, timestamp, signature, expected_signature
+        );
+
+        return Err(std::io::Error::other("Invalid signature"));
+    }
+
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::{HeaderMap, Method};
+    use time::OffsetDateTime;
+
+    #[test]
+    fn test_get_shared_secret() {
+        let secret = get_shared_secret();
+        assert!(!secret.is_empty(), "Secret should not be empty");
+
+        let url = "http://node1:7000/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A7000%2Fdata%2Frustfs3&volume=.rustfs.sys&path=pool.bin%2Fdd0fd773-a962-4265-b543-783ce83953e9%2Fpart.1&offset=0&length=44";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        build_auth_headers(url, &method, &mut headers);
+
+        let url = "/rustfs/rpc/read_file_stream?disk=http%3A%2F%2Fnode1%3A7000%2Fdata%2Frustfs3&volume=.rustfs.sys&path=pool.bin%2Fdd0fd773-a962-4265-b543-783ce83953e9%2Fpart.1&offset=0&length=44";
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_ok(), "Valid signature should pass verification");
+    }
+
+    #[test]
+    fn test_generate_signature_deterministic() {
+        let secret = "test-secret";
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let timestamp = 1640995200; // Fixed timestamp
+
+        let signature1 = generate_signature(secret, url, &method, timestamp);
+        let signature2 = generate_signature(secret, url, &method, timestamp);
+
+        assert_eq!(signature1, signature2, "Same inputs should produce same signature");
+        assert!(!signature1.is_empty(), "Signature should not be empty");
+    }
+
+    #[test]
+    fn test_generate_signature_different_inputs() {
+        let secret = "test-secret";
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let timestamp = 1640995200;
+
+        let signature1 = generate_signature(secret, url, &method, timestamp);
+        let signature2 = generate_signature(secret, "http://different.com/api/test2", &method, timestamp);
+        let signature3 = generate_signature(secret, url, &Method::POST, timestamp);
+        let signature4 = generate_signature(secret, url, &method, timestamp + 1);
+
+        assert_ne!(signature1, signature2, "Different URLs should produce different signatures");
+        assert_ne!(signature1, signature3, "Different methods should produce different signatures");
+        assert_ne!(signature1, signature4, "Different timestamps should produce different signatures");
+    }
+
+    #[test]
+    fn test_build_auth_headers() {
+        let url = "http://example.com/api/test";
+        let method = Method::POST;
+        let mut headers = HeaderMap::new();
+
+        build_auth_headers(url, &method, &mut headers);
+
+        // Verify headers are present
+        assert!(headers.contains_key(SIGNATURE_HEADER), "Should contain signature header");
+        assert!(headers.contains_key(TIMESTAMP_HEADER), "Should contain timestamp header");
+
+        // Verify header values are not empty
+        let signature = headers.get(SIGNATURE_HEADER).unwrap().to_str().unwrap();
+        let timestamp_str = headers.get(TIMESTAMP_HEADER).unwrap().to_str().unwrap();
+
+        assert!(!signature.is_empty(), "Signature should not be empty");
+        assert!(!timestamp_str.is_empty(), "Timestamp should not be empty");
+
+        // Verify timestamp is a valid integer
+        let timestamp: i64 = timestamp_str.parse().expect("Timestamp should be valid integer");
+        let current_time = OffsetDateTime::now_utc().unix_timestamp();
+
+        // Should be within a reasonable range (within 1 second of current time)
+        assert!((current_time - timestamp).abs() <= 1, "Timestamp should be close to current time");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_success() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Build headers with valid signature
+        build_auth_headers(url, &method, &mut headers);
+
+        // Verify should succeed
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_ok(), "Valid signature should pass verification");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_invalid_signature() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Build headers with valid signature first
+        build_auth_headers(url, &method, &mut headers);
+
+        // Tamper with the signature
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str("invalid-signature").unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Invalid signature should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid signature");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_expired_timestamp() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Set expired timestamp (older than SIGNATURE_VALID_DURATION)
+        let expired_timestamp = OffsetDateTime::now_utc().unix_timestamp() - SIGNATURE_VALID_DURATION - 10;
+        let secret = get_shared_secret();
+        let signature = generate_signature(&secret, url, &method, expired_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&expired_timestamp.to_string()).unwrap());
+
+        // Verify should fail due to expired timestamp
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Expired timestamp should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Request timestamp expired");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_missing_signature_header() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Add only timestamp header, missing signature
+        let timestamp = OffsetDateTime::now_utc().unix_timestamp();
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&timestamp.to_string()).unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Missing signature header should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Missing signature header");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_missing_timestamp_header() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Add only signature header, missing timestamp
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str("some-signature").unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Missing timestamp header should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Missing timestamp header");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_invalid_timestamp_format() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str("some-signature").unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str("invalid-timestamp").unwrap());
+
+        // Verify should fail
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Invalid timestamp format should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid timestamp format");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_url_mismatch() {
+        let original_url = "http://example.com/api/test";
+        let different_url = "http://example.com/api/different";
+        let method = Method::GET;
+        let mut headers = HeaderMap::new();
+
+        // Build headers for one URL
+        build_auth_headers(original_url, &method, &mut headers);
+
+        // Try to verify with a different URL
+        let result = verify_rpc_signature(different_url, &method, &headers);
+        assert!(result.is_err(), "URL mismatch should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid signature");
+    }
+
+    #[test]
+    fn test_verify_rpc_signature_method_mismatch() {
+        let url = "http://example.com/api/test";
+        let original_method = Method::GET;
+        let different_method = Method::POST;
+        let mut headers = HeaderMap::new();
+
+        // Build headers for one method
+        build_auth_headers(url, &original_method, &mut headers);
+
+        // Try to verify with a different method
+        let result = verify_rpc_signature(url, &different_method, &headers);
+        assert!(result.is_err(), "Method mismatch should fail verification");
+
+        let error = result.unwrap_err();
+        assert_eq!(error.to_string(), "Invalid signature");
+    }
+
+    #[test]
+    fn test_signature_valid_duration_boundary() {
+        let url = "http://example.com/api/test";
+        let method = Method::GET;
+        let secret = get_shared_secret();
+
+        let mut headers = HeaderMap::new();
+        let current_time = OffsetDateTime::now_utc().unix_timestamp();
+        // Test timestamp just within valid duration
+        let valid_timestamp = current_time - SIGNATURE_VALID_DURATION + 1;
+
+        let signature = generate_signature(&secret, url, &method, valid_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&valid_timestamp.to_string()).unwrap());
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_ok(), "Timestamp within valid duration should pass");
+
+        // Test timestamp just outside valid duration
+        let mut headers = HeaderMap::new();
+        let invalid_timestamp = current_time - SIGNATURE_VALID_DURATION - 15;
+        let signature = generate_signature(&secret, url, &method, invalid_timestamp);
+
+        headers.insert(SIGNATURE_HEADER, HeaderValue::from_str(&signature).unwrap());
+        headers.insert(TIMESTAMP_HEADER, HeaderValue::from_str(&invalid_timestamp.to_string()).unwrap());
+
+        let result = verify_rpc_signature(url, &method, &headers);
+        assert!(result.is_err(), "Timestamp outside valid duration should fail");
+    }
+
+    #[test]
+    fn test_round_trip_authentication() {
+        let test_cases = vec![
+            ("http://example.com/api/test", Method::GET),
+            ("https://api.rustfs.com/v1/bucket", Method::POST),
+            ("http://localhost:9000/admin/info", Method::PUT),
+            ("https://storage.example.com/path/to/object?query=param", Method::DELETE),
+        ];
+
+        for (url, method) in test_cases {
+            let mut headers = HeaderMap::new();
+
+            // Build authentication headers
+            build_auth_headers(url, &method, &mut headers);
+
+            // Verify the signature should succeed
+            let result = verify_rpc_signature(url, &method, &headers);
+            assert!(result.is_ok(), "Round-trip test failed for {} {}", method, url);
+        }
+    }
+}

ecstore/src/rpc/mod.rs

@@ -0,0 +1,11 @@
+mod http_auth;
+mod peer_rest_client;
+mod peer_s3_client;
+mod remote_disk;
+mod tonic_service;
+
+pub use http_auth::{build_auth_headers, verify_rpc_signature};
+pub use peer_rest_client::PeerRestClient;
+pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, RemotePeerS3Client, S3PeerSys};
+pub use remote_disk::RemoteDisk;
+pub use tonic_service::make_server;

ecstore/src/rpc/peer_rest_client.rs

@@ -0,0 +1,694 @@
+use crate::error::{Error, Result};
+use crate::{
+    endpoints::EndpointServerPools,
+    global::is_dist_erasure,
+    heal::heal_commands::BgHealState,
+    metrics_realtime::{CollectMetricsOpts, MetricType},
+};
+use madmin::{
+    ServerProperties,
+    health::{Cpus, MemInfo, OsInfo, Partitions, ProcInfo, SysConfig, SysErrors, SysService},
+    metrics::RealtimeMetrics,
+    net::NetInfo,
+};
+use protos::{
+    node_service_time_out_client,
+    proto_gen::node_service::{
+        BackgroundHealStatusRequest, DeleteBucketMetadataRequest, DeletePolicyRequest, DeleteServiceAccountRequest,
+        DeleteUserRequest, GetCpusRequest, GetMemInfoRequest, GetMetricsRequest, GetNetInfoRequest, GetOsInfoRequest,
+        GetPartitionsRequest, GetProcInfoRequest, GetSeLinuxInfoRequest, GetSysConfigRequest, GetSysErrorsRequest,
+        LoadBucketMetadataRequest, LoadGroupRequest, LoadPolicyMappingRequest, LoadPolicyRequest, LoadRebalanceMetaRequest,
+        LoadServiceAccountRequest, LoadTransitionTierConfigRequest, LoadUserRequest, LocalStorageInfoRequest, Mss,
+        ReloadPoolMetaRequest, ReloadSiteReplicationConfigRequest, ServerInfoRequest, SignalServiceRequest,
+        StartProfilingRequest, StopRebalanceRequest,
+    },
+};
+use rmp_serde::{Deserializer, Serializer};
+use rustfs_utils::XHost;
+use serde::{Deserialize, Serialize as _};
+use std::{collections::HashMap, io::Cursor, time::SystemTime};
+use tonic::Request;
+use tracing::warn;
+
+pub const PEER_RESTSIGNAL: &str = "signal";
+pub const PEER_RESTSUB_SYS: &str = "sub-sys";
+pub const PEER_RESTDRY_RUN: &str = "dry-run";
+
+#[derive(Clone, Debug)]
+pub struct PeerRestClient {
+    pub host: XHost,
+    pub grid_host: String,
+}
+
+impl PeerRestClient {
+    pub fn new(host: XHost, grid_host: String) -> Self {
+        Self { host, grid_host }
+    }
+    pub async fn new_clients(eps: EndpointServerPools) -> (Vec<Option<Self>>, Vec<Option<Self>>) {
+        if !is_dist_erasure().await {
+            return (Vec::new(), Vec::new());
+        }
+
+        let eps = eps.clone();
+        let hosts = eps.hosts_sorted();
+        let mut remote = Vec::with_capacity(hosts.len());
+        let mut all = vec![None; hosts.len()];
+        for (i, hs_host) in hosts.iter().enumerate() {
+            if let Some(host) = hs_host {
+                if let Some(grid_host) = eps.find_grid_hosts_from_peer(host) {
+                    let client = PeerRestClient::new(host.clone(), grid_host);
+
+                    all[i] = Some(client.clone());
+                    remote.push(Some(client));
+                }
+            }
+        }
+
+        if all.len() != remote.len() + 1 {
+            warn!("Expected number of all hosts ({}) to be remote +1 ({})", all.len(), remote.len());
+        }
+
+        (remote, all)
+    }
+}
+
+impl PeerRestClient {
+    pub async fn local_storage_info(&self) -> Result<madmin::StorageInfo> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LocalStorageInfoRequest { metrics: true });
+
+        let response = client.local_storage_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.storage_info;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let storage_info: madmin::StorageInfo = Deserialize::deserialize(&mut buf)?;
+
+        Ok(storage_info)
+    }
+
+    pub async fn server_info(&self) -> Result<ServerProperties> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(ServerInfoRequest { metrics: true });
+
+        let response = client.server_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.server_properties;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let storage_properties: ServerProperties = Deserialize::deserialize(&mut buf)?;
+
+        Ok(storage_properties)
+    }
+
+    pub async fn get_cpus(&self) -> Result<Cpus> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetCpusRequest {});
+
+        let response = client.get_cpus(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.cpus;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let cpus: Cpus = Deserialize::deserialize(&mut buf)?;
+
+        Ok(cpus)
+    }
+
+    pub async fn get_net_info(&self) -> Result<NetInfo> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetNetInfoRequest {});
+
+        let response = client.get_net_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.net_info;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let net_info: NetInfo = Deserialize::deserialize(&mut buf)?;
+
+        Ok(net_info)
+    }
+
+    pub async fn get_partitions(&self) -> Result<Partitions> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetPartitionsRequest {});
+
+        let response = client.get_partitions(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.partitions;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let partitions: Partitions = Deserialize::deserialize(&mut buf)?;
+
+        Ok(partitions)
+    }
+
+    pub async fn get_os_info(&self) -> Result<OsInfo> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetOsInfoRequest {});
+
+        let response = client.get_os_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.os_info;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let os_info: OsInfo = Deserialize::deserialize(&mut buf)?;
+
+        Ok(os_info)
+    }
+
+    pub async fn get_se_linux_info(&self) -> Result<SysService> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetSeLinuxInfoRequest {});
+
+        let response = client.get_se_linux_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.sys_services;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let sys_services: SysService = Deserialize::deserialize(&mut buf)?;
+
+        Ok(sys_services)
+    }
+
+    pub async fn get_sys_config(&self) -> Result<SysConfig> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetSysConfigRequest {});
+
+        let response = client.get_sys_config(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.sys_config;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let sys_config: SysConfig = Deserialize::deserialize(&mut buf)?;
+
+        Ok(sys_config)
+    }
+
+    pub async fn get_sys_errors(&self) -> Result<SysErrors> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetSysErrorsRequest {});
+
+        let response = client.get_sys_errors(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.sys_errors;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let sys_errors: SysErrors = Deserialize::deserialize(&mut buf)?;
+
+        Ok(sys_errors)
+    }
+
+    pub async fn get_mem_info(&self) -> Result<MemInfo> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetMemInfoRequest {});
+
+        let response = client.get_mem_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.mem_info;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let mem_info: MemInfo = Deserialize::deserialize(&mut buf)?;
+
+        Ok(mem_info)
+    }
+
+    pub async fn get_metrics(&self, t: MetricType, opts: &CollectMetricsOpts) -> Result<RealtimeMetrics> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let mut buf_t = Vec::new();
+        t.serialize(&mut Serializer::new(&mut buf_t))?;
+        let mut buf_o = Vec::new();
+        opts.serialize(&mut Serializer::new(&mut buf_o))?;
+        let request = Request::new(GetMetricsRequest {
+            metric_type: buf_t.into(),
+            opts: buf_o.into(),
+        });
+
+        let response = client.get_metrics(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.realtime_metrics;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let realtime_metrics: RealtimeMetrics = Deserialize::deserialize(&mut buf)?;
+
+        Ok(realtime_metrics)
+    }
+
+    pub async fn get_proc_info(&self) -> Result<ProcInfo> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(GetProcInfoRequest {});
+
+        let response = client.get_proc_info(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.proc_info;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let proc_info: ProcInfo = Deserialize::deserialize(&mut buf)?;
+
+        Ok(proc_info)
+    }
+
+    pub async fn start_profiling(&self, profiler: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(StartProfilingRequest {
+            profiler: profiler.to_string(),
+        });
+
+        let response = client.start_profiling(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn download_profile_data(&self) -> Result<()> {
+        todo!()
+    }
+
+    pub async fn get_bucket_stats(&self) -> Result<()> {
+        todo!()
+    }
+
+    pub async fn get_sr_metrics(&self) -> Result<()> {
+        todo!()
+    }
+
+    pub async fn get_all_bucket_stats(&self) -> Result<()> {
+        todo!()
+    }
+
+    pub async fn load_bucket_metadata(&self, bucket: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadBucketMetadataRequest {
+            bucket: bucket.to_string(),
+        });
+
+        let response = client.load_bucket_metadata(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn delete_bucket_metadata(&self, bucket: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(DeleteBucketMetadataRequest {
+            bucket: bucket.to_string(),
+        });
+
+        let response = client.delete_bucket_metadata(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn delete_policy(&self, policy: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(DeletePolicyRequest {
+            policy_name: policy.to_string(),
+        });
+
+        let response = client.delete_policy(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn load_policy(&self, policy: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadPolicyRequest {
+            policy_name: policy.to_string(),
+        });
+
+        let response = client.load_policy(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn load_policy_mapping(&self, user_or_group: &str, user_type: u64, is_group: bool) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadPolicyMappingRequest {
+            user_or_group: user_or_group.to_string(),
+            user_type,
+            is_group,
+        });
+
+        let response = client.load_policy_mapping(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn delete_user(&self, access_key: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(DeleteUserRequest {
+            access_key: access_key.to_string(),
+        });
+
+        let response = client.delete_user(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn delete_service_account(&self, access_key: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(DeleteServiceAccountRequest {
+            access_key: access_key.to_string(),
+        });
+
+        let response = client.delete_service_account(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn load_user(&self, access_key: &str, temp: bool) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadUserRequest {
+            access_key: access_key.to_string(),
+            temp,
+        });
+
+        let response = client.load_user(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn load_service_account(&self, access_key: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadServiceAccountRequest {
+            access_key: access_key.to_string(),
+        });
+
+        let response = client.load_service_account(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn load_group(&self, group: &str) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadGroupRequest {
+            group: group.to_string(),
+        });
+
+        let response = client.load_group(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn reload_site_replication_config(&self) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(ReloadSiteReplicationConfigRequest {});
+
+        let response = client.reload_site_replication_config(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn signal_service(&self, sig: u64, sub_sys: &str, dry_run: bool, _exec_at: SystemTime) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let mut vars = HashMap::new();
+        vars.insert(PEER_RESTSIGNAL.to_string(), sig.to_string());
+        vars.insert(PEER_RESTSUB_SYS.to_string(), sub_sys.to_string());
+        vars.insert(PEER_RESTDRY_RUN.to_string(), dry_run.to_string());
+        let request = Request::new(SignalServiceRequest {
+            vars: Some(Mss { value: vars }),
+        });
+
+        let response = client.signal_service(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        Ok(())
+    }
+
+    pub async fn background_heal_status(&self) -> Result<BgHealState> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(BackgroundHealStatusRequest {});
+
+        let response = client.background_heal_status(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+        let data = response.bg_heal_state;
+
+        let mut buf = Deserializer::new(Cursor::new(data));
+        let bg_heal_state: BgHealState = Deserialize::deserialize(&mut buf)?;
+
+        Ok(bg_heal_state)
+    }
+
+    pub async fn get_metacache_listing(&self) -> Result<()> {
+        let _client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        todo!()
+    }
+
+    pub async fn update_metacache_listing(&self) -> Result<()> {
+        let _client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        todo!()
+    }
+
+    pub async fn reload_pool_meta(&self) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(ReloadPoolMetaRequest {});
+
+        let response = client.reload_pool_meta(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+
+        Ok(())
+    }
+
+    pub async fn stop_rebalance(&self) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(StopRebalanceRequest {});
+
+        let response = client.stop_rebalance(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+
+        Ok(())
+    }
+
+    pub async fn load_rebalance_meta(&self, start_rebalance: bool) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadRebalanceMetaRequest { start_rebalance });
+
+        let response = client.load_rebalance_meta(request).await?.into_inner();
+
+        warn!("load_rebalance_meta response {:?}, grid_host: {:?}", response, &self.grid_host);
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+
+        Ok(())
+    }
+
+    pub async fn load_transition_tier_config(&self) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.grid_host)
+            .await
+            .map_err(|err| Error::other(err.to_string()))?;
+        let request = Request::new(LoadTransitionTierConfigRequest {});
+
+        let response = client.load_transition_tier_config(request).await?.into_inner();
+        if !response.success {
+            if let Some(msg) = response.error_info {
+                return Err(Error::other(msg));
+            }
+            return Err(Error::other(""));
+        }
+
+        Ok(())
+    }
+}

ecstore/src/rpc/peer_s3_client.rs

@@ -0,0 +1,757 @@
+use crate::bucket::metadata_sys;
+use crate::disk::error::{Error, Result};
+use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs};
+use crate::disk::{DiskAPI, DiskStore};
+use crate::global::GLOBAL_LOCAL_DISK_MAP;
+use crate::heal::heal_commands::{
+    DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET, HealOpts,
+};
+use crate::heal::heal_ops::RUSTFS_RESERVED_BUCKET;
+use crate::store::all_local_disk;
+use crate::store_utils::is_reserved_or_invalid_bucket;
+use crate::{
+    disk::{self, VolumeInfo},
+    endpoints::{EndpointServerPools, Node},
+    store_api::{BucketInfo, BucketOptions, DeleteBucketOptions, MakeBucketOptions},
+};
+use async_trait::async_trait;
+use futures::future::join_all;
+use madmin::heal_commands::{HealDriveInfo, HealResultItem};
+use protos::node_service_time_out_client;
+use protos::proto_gen::node_service::{
+    DeleteBucketRequest, GetBucketInfoRequest, HealBucketRequest, ListBucketRequest, MakeBucketRequest,
+};
+use std::{collections::HashMap, fmt::Debug, sync::Arc};
+use tokio::sync::RwLock;
+use tonic::Request;
+use tracing::info;
+
+type Client = Arc<Box<dyn PeerS3Client>>;
+
+#[async_trait]
+pub trait PeerS3Client: Debug + Sync + Send + 'static {
+    async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem>;
+    async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
+    async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
+    async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
+    async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
+    fn get_pools(&self) -> Option<Vec<usize>>;
+}
+
+#[derive(Debug, Clone)]
+pub struct S3PeerSys {
+    pub clients: Vec<Client>,
+    pub pools_count: usize,
+}
+
+impl S3PeerSys {
+    pub fn new(eps: &EndpointServerPools) -> Self {
+        Self {
+            clients: Self::new_clients(eps),
+            pools_count: eps.as_ref().len(),
+        }
+    }
+
+    fn new_clients(eps: &EndpointServerPools) -> Vec<Client> {
+        let nodes = eps.get_nodes();
+        let v: Vec<Client> = nodes
+            .iter()
+            .map(|e| {
+                if e.is_local {
+                    let cli: Box<dyn PeerS3Client> = Box::new(LocalPeerS3Client::new(Some(e.clone()), Some(e.pools.clone())));
+                    Arc::new(cli)
+                } else {
+                    let cli: Box<dyn PeerS3Client> = Box::new(RemotePeerS3Client::new(Some(e.clone()), Some(e.pools.clone())));
+                    Arc::new(cli)
+                }
+            })
+            .collect();
+
+        v
+    }
+}
+
+impl S3PeerSys {
+    pub async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
+        let mut opts = *opts;
+        let mut futures = Vec::with_capacity(self.clients.len());
+        for client in self.clients.iter() {
+            // client_clon
+            futures.push(async move { (client.get_bucket_info(bucket, &BucketOptions::default()).await).err() });
+        }
+        let errs = join_all(futures).await;
+
+        let mut pool_errs = Vec::new();
+        for pool_idx in 0..self.pools_count {
+            let mut per_pool_errs = Vec::new();
+            for (i, client) in self.clients.iter().enumerate() {
+                if let Some(v) = client.get_pools() {
+                    if v.contains(&pool_idx) {
+                        per_pool_errs.push(errs[i].clone());
+                    }
+                }
+            }
+            let qu = per_pool_errs.len() / 2;
+            pool_errs.push(reduce_write_quorum_errs(&per_pool_errs, BUCKET_OP_IGNORED_ERRS, qu));
+        }
+
+        if !opts.recreate {
+            opts.remove = is_all_buckets_not_found(&pool_errs);
+            opts.recreate = !opts.remove;
+        }
+
+        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;
+            let heal_bucket_results_clone = heal_bucket_results.clone();
+            futures.push(async move {
+                match client.heal_bucket(bucket, &opts_clone).await {
+                    Ok(res) => {
+                        heal_bucket_results_clone.write().await[idx] = res;
+                        None
+                    }
+                    Err(err) => Some(err),
+                }
+            });
+        }
+        let errs = join_all(futures).await;
+
+        for pool_idx in 0..self.pools_count {
+            let mut per_pool_errs = Vec::new();
+            for (i, client) in self.clients.iter().enumerate() {
+                if let Some(v) = client.get_pools() {
+                    if v.contains(&pool_idx) {
+                        per_pool_errs.push(errs[i].clone());
+                    }
+                }
+            }
+            let qu = per_pool_errs.len() / 2;
+            if let Some(pool_err) = reduce_write_quorum_errs(&per_pool_errs, BUCKET_OP_IGNORED_ERRS, qu) {
+                return Err(pool_err);
+            }
+        }
+
+        for (i, err) in errs.iter().enumerate() {
+            if err.is_none() {
+                return Ok(heal_bucket_results.read().await[i].clone());
+            }
+        }
+        Err(Error::VolumeNotFound)
+    }
+
+    pub async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
+        let mut futures = Vec::with_capacity(self.clients.len());
+        for cli in self.clients.iter() {
+            futures.push(cli.make_bucket(bucket, opts));
+        }
+
+        let mut errors = Vec::with_capacity(self.clients.len());
+
+        let results = join_all(futures).await;
+        for result in results {
+            match result {
+                Ok(_) => {
+                    errors.push(None);
+                }
+                Err(e) => {
+                    errors.push(Some(e));
+                }
+            }
+        }
+
+        for i in 0..self.pools_count {
+            let mut per_pool_errs = Vec::with_capacity(self.clients.len());
+            for (j, cli) in self.clients.iter().enumerate() {
+                let pools = cli.get_pools();
+                let idx = i;
+                if pools.unwrap_or_default().contains(&idx) {
+                    per_pool_errs.push(errors[j].as_ref());
+                }
+
+                // TODO: reduceWriteQuorumErrs
+            }
+        }
+
+        // TODO:
+
+        Ok(())
+    }
+    pub async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
+        let mut futures = Vec::with_capacity(self.clients.len());
+        for cli in self.clients.iter() {
+            futures.push(cli.list_bucket(opts));
+        }
+
+        let mut errors = Vec::with_capacity(self.clients.len());
+        let mut ress = Vec::with_capacity(self.clients.len());
+
+        let results = join_all(futures).await;
+        for result in results {
+            match result {
+                Ok(res) => {
+                    ress.push(Some(res));
+                    errors.push(None);
+                }
+                Err(e) => {
+                    ress.push(None);
+                    errors.push(Some(e));
+                }
+            }
+        }
+        // TODO: reduceWriteQuorumErrs
+        // for i in 0..self.pools_count {}
+
+        let mut uniq_map: HashMap<&String, &BucketInfo> = HashMap::new();
+
+        for res in ress.iter() {
+            if res.is_none() {
+                continue;
+            }
+
+            let buckets = res.as_ref().unwrap();
+
+            for bucket in buckets.iter() {
+                if !uniq_map.contains_key(&bucket.name) {
+                    uniq_map.insert(&bucket.name, bucket);
+                }
+            }
+        }
+
+        let buckets: Vec<BucketInfo> = uniq_map.values().map(|&v| v.clone()).collect();
+
+        Ok(buckets)
+    }
+    pub async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> {
+        let mut futures = Vec::with_capacity(self.clients.len());
+        for cli in self.clients.iter() {
+            futures.push(cli.delete_bucket(bucket, opts));
+        }
+
+        let mut errors = Vec::with_capacity(self.clients.len());
+
+        let results = join_all(futures).await;
+
+        for result in results {
+            match result {
+                Ok(_) => {
+                    errors.push(None);
+                }
+                Err(e) => {
+                    errors.push(Some(e));
+                }
+            }
+        }
+
+        // TODO: reduceWriteQuorumErrs
+
+        Ok(())
+    }
+    pub async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
+        let mut futures = Vec::with_capacity(self.clients.len());
+        for cli in self.clients.iter() {
+            futures.push(cli.get_bucket_info(bucket, opts));
+        }
+
+        let mut ress = Vec::with_capacity(self.clients.len());
+        let mut errors = Vec::with_capacity(self.clients.len());
+
+        let results = join_all(futures).await;
+        for result in results {
+            match result {
+                Ok(res) => {
+                    ress.push(Some(res));
+                    errors.push(None);
+                }
+                Err(e) => {
+                    ress.push(None);
+                    errors.push(Some(e));
+                }
+            }
+        }
+
+        for i in 0..self.pools_count {
+            let mut per_pool_errs = Vec::with_capacity(self.clients.len());
+            for (j, cli) in self.clients.iter().enumerate() {
+                let pools = cli.get_pools();
+                let idx = i;
+                if pools.unwrap_or_default().contains(&idx) {
+                    per_pool_errs.push(errors[j].as_ref());
+                }
+
+                // TODO: reduceWriteQuorumErrs
+            }
+        }
+
+        ress.iter().find_map(|op| op.clone()).ok_or(Error::VolumeNotFound)
+    }
+
+    pub fn get_pools(&self) -> Option<Vec<usize>> {
+        unimplemented!()
+    }
+}
+
+#[derive(Debug)]
+pub struct LocalPeerS3Client {
+    // pub local_disks: Vec<DiskStore>,
+    // pub node: Node,
+    pub pools: Option<Vec<usize>>,
+}
+
+impl LocalPeerS3Client {
+    pub fn new(_node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
+        Self {
+            // local_disks,
+            // node,
+            pools,
+        }
+    }
+}
+
+#[async_trait]
+impl PeerS3Client for LocalPeerS3Client {
+    fn get_pools(&self) -> Option<Vec<usize>> {
+        self.pools.clone()
+    }
+
+    async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
+        heal_bucket_local(bucket, opts).await
+    }
+
+    async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
+        let local_disks = all_local_disk().await;
+
+        let mut futures = Vec::with_capacity(local_disks.len());
+        for disk in local_disks.iter() {
+            futures.push(disk.list_volumes());
+        }
+
+        let results = join_all(futures).await;
+
+        let mut ress = Vec::new();
+        let mut errs = Vec::new();
+
+        for result in results {
+            match result {
+                Ok(res) => {
+                    ress.push(res);
+                    errs.push(None);
+                }
+                Err(e) => errs.push(Some(e)),
+            }
+        }
+
+        let mut uniq_map: HashMap<&String, &VolumeInfo> = HashMap::new();
+
+        for info_list in ress.iter() {
+            for info in info_list.iter() {
+                if is_reserved_or_invalid_bucket(&info.name, false) {
+                    continue;
+                }
+                if !uniq_map.contains_key(&info.name) {
+                    uniq_map.insert(&info.name, info);
+                }
+            }
+        }
+
+        let buckets: Vec<BucketInfo> = uniq_map
+            .values()
+            .map(|&v| BucketInfo {
+                name: v.name.clone(),
+                created: v.created,
+                ..Default::default()
+            })
+            .collect();
+
+        Ok(buckets)
+    }
+    async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
+        let local_disks = all_local_disk().await;
+        let mut futures = Vec::with_capacity(local_disks.len());
+        for disk in local_disks.iter() {
+            futures.push(async move {
+                match disk.make_volume(bucket).await {
+                    Ok(_) => Ok(()),
+                    Err(e) => {
+                        if opts.force_create && matches!(e, Error::VolumeExists) {
+                            return Ok(());
+                        }
+
+                        Err(e)
+                    }
+                }
+            });
+        }
+
+        let results = join_all(futures).await;
+
+        let mut errs = Vec::new();
+
+        for res in results {
+            match res {
+                Ok(_) => errs.push(None),
+                Err(e) => errs.push(Some(e)),
+            }
+        }
+
+        // TODO: reduceWriteQuorumErrs
+
+        Ok(())
+    }
+
+    async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
+        let local_disks = all_local_disk().await;
+        let mut futures = Vec::with_capacity(local_disks.len());
+        for disk in local_disks.iter() {
+            futures.push(disk.stat_volume(bucket));
+        }
+
+        let results = join_all(futures).await;
+
+        let mut ress = Vec::with_capacity(local_disks.len());
+        let mut errs = Vec::with_capacity(local_disks.len());
+
+        for res in results {
+            match res {
+                Ok(r) => {
+                    errs.push(None);
+                    ress.push(Some(r));
+                }
+                Err(e) => {
+                    errs.push(Some(e));
+                    ress.push(None);
+                }
+            }
+        }
+
+        // TODO: reduceWriteQuorumErrs
+        let mut versioned = false;
+        if let Ok(sys) = metadata_sys::get(bucket).await {
+            versioned = sys.versioning();
+        }
+
+        ress.iter()
+            .find_map(|op| {
+                op.as_ref().map(|v| BucketInfo {
+                    name: v.name.clone(),
+                    created: v.created,
+                    versionning: versioned,
+                    ..Default::default()
+                })
+            })
+            .ok_or(Error::VolumeNotFound)
+    }
+
+    async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
+        let local_disks = all_local_disk().await;
+        let mut futures = Vec::with_capacity(local_disks.len());
+
+        for disk in local_disks.iter() {
+            futures.push(disk.delete_volume(bucket));
+        }
+
+        let results = join_all(futures).await;
+
+        let mut errs = Vec::new();
+
+        let mut recreate = false;
+
+        for res in results {
+            match res {
+                Ok(_) => errs.push(None),
+                Err(e) => {
+                    if matches!(e, Error::VolumeNotEmpty) {
+                        recreate = true;
+                    }
+                    errs.push(Some(e))
+                }
+            }
+        }
+
+        // errVolumeNotEmpty 不删除，把已经删除的重新创建
+
+        for (idx, err) in errs.into_iter().enumerate() {
+            if err.is_none() && recreate {
+                let _ = local_disks[idx].make_volume(bucket).await;
+            }
+        }
+
+        if recreate {
+            return Err(Error::VolumeNotEmpty);
+        }
+
+        // TODO: reduceWriteQuorumErrs
+
+        Ok(())
+    }
+}
+
+#[derive(Debug)]
+pub struct RemotePeerS3Client {
+    pub node: Option<Node>,
+    pub pools: Option<Vec<usize>>,
+    addr: String,
+}
+
+impl RemotePeerS3Client {
+    pub fn new(node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
+        let addr = node.as_ref().map(|v| v.url.to_string()).unwrap_or_default().to_string();
+        Self { node, pools, addr }
+    }
+    pub fn get_addr(&self) -> String {
+        self.addr.clone()
+    }
+}
+
+#[async_trait]
+impl PeerS3Client for RemotePeerS3Client {
+    fn get_pools(&self) -> Option<Vec<usize>> {
+        self.pools.clone()
+    }
+
+    async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
+        let options: String = serde_json::to_string(opts)?;
+        let mut client = node_service_time_out_client(&self.addr)
+            .await
+            .map_err(|err| Error::other(format!("can not get client, err: {}", err)))?;
+        let request = Request::new(HealBucketRequest {
+            bucket: bucket.to_string(),
+            options,
+        });
+        let response = client.heal_bucket(request).await?.into_inner();
+        if !response.success {
+            return if let Some(err) = response.error {
+                Err(err.into())
+            } else {
+                Err(Error::other(""))
+            };
+        }
+
+        Ok(HealResultItem {
+            heal_item_type: HEAL_ITEM_BUCKET.to_string(),
+            bucket: bucket.to_string(),
+            set_count: 0,
+            ..Default::default()
+        })
+    }
+
+    async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
+        let options = serde_json::to_string(opts)?;
+        let mut client = node_service_time_out_client(&self.addr)
+            .await
+            .map_err(|err| Error::other(format!("can not get client, err: {}", err)))?;
+        let request = Request::new(ListBucketRequest { options });
+        let response = client.list_bucket(request).await?.into_inner();
+        if !response.success {
+            return if let Some(err) = response.error {
+                Err(err.into())
+            } else {
+                Err(Error::other(""))
+            };
+        }
+        let bucket_infos = response
+            .bucket_infos
+            .into_iter()
+            .filter_map(|json_str| serde_json::from_str::<BucketInfo>(&json_str).ok())
+            .collect();
+
+        Ok(bucket_infos)
+    }
+    async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
+        let options = serde_json::to_string(opts)?;
+        let mut client = node_service_time_out_client(&self.addr)
+            .await
+            .map_err(|err| Error::other(format!("can not get client, err: {}", err)))?;
+        let request = Request::new(MakeBucketRequest {
+            name: bucket.to_string(),
+            options,
+        });
+        let response = client.make_bucket(request).await?.into_inner();
+
+        // TODO: deal with error
+        if !response.success {
+            return if let Some(err) = response.error {
+                Err(err.into())
+            } else {
+                Err(Error::other(""))
+            };
+        }
+
+        Ok(())
+    }
+    async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> {
+        let options = serde_json::to_string(opts)?;
+        let mut client = node_service_time_out_client(&self.addr)
+            .await
+            .map_err(|err| Error::other(format!("can not get client, err: {}", err)))?;
+        let request = Request::new(GetBucketInfoRequest {
+            bucket: bucket.to_string(),
+            options,
+        });
+        let response = client.get_bucket_info(request).await?.into_inner();
+        if !response.success {
+            return if let Some(err) = response.error {
+                Err(err.into())
+            } else {
+                Err(Error::other(""))
+            };
+        }
+        let bucket_info = serde_json::from_str::<BucketInfo>(&response.bucket_info)?;
+
+        Ok(bucket_info)
+    }
+
+    async fn delete_bucket(&self, bucket: &str, _opts: &DeleteBucketOptions) -> Result<()> {
+        let mut client = node_service_time_out_client(&self.addr)
+            .await
+            .map_err(|err| Error::other(format!("can not get client, err: {}", err)))?;
+
+        let request = Request::new(DeleteBucketRequest {
+            bucket: bucket.to_string(),
+        });
+        let response = client.delete_bucket(request).await?.into_inner();
+        if !response.success {
+            return if let Some(err) = response.error {
+                Err(err.into())
+            } else {
+                Err(Error::other(""))
+            };
+        }
+
+        Ok(())
+    }
+}
+
+pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
+    let disks = clone_drives().await;
+    let before_state = Arc::new(RwLock::new(vec![String::new(); disks.len()]));
+    let after_state = Arc::new(RwLock::new(vec![String::new(); disks.len()]));
+
+    let mut futures = Vec::new();
+    for (index, disk) in disks.iter().enumerate() {
+        let disk = disk.clone();
+        let bucket = bucket.to_string();
+        let bs_clone = before_state.clone();
+        let as_clone = after_state.clone();
+        futures.push(async move {
+            let disk = match disk {
+                Some(disk) => disk,
+                None => {
+                    bs_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string();
+                    as_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string();
+                    return Some(Error::DiskNotFound);
+                }
+            };
+            bs_clone.write().await[index] = DRIVE_STATE_OK.to_string();
+            as_clone.write().await[index] = DRIVE_STATE_OK.to_string();
+
+            if bucket == RUSTFS_RESERVED_BUCKET {
+                return None;
+            }
+
+            match disk.stat_volume(&bucket).await {
+                Ok(_) => None,
+                Err(err) => match err {
+                    Error::DiskNotFound => {
+                        bs_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string();
+                        as_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string();
+                        Some(err)
+                    }
+                    Error::VolumeNotFound => {
+                        bs_clone.write().await[index] = DRIVE_STATE_MISSING.to_string();
+                        as_clone.write().await[index] = DRIVE_STATE_MISSING.to_string();
+                        Some(err)
+                    }
+                    _ => {
+                        bs_clone.write().await[index] = DRIVE_STATE_CORRUPT.to_string();
+                        as_clone.write().await[index] = DRIVE_STATE_CORRUPT.to_string();
+                        Some(err)
+                    }
+                },
+            }
+        });
+    }
+    let errs = join_all(futures).await;
+    let mut res = HealResultItem {
+        heal_item_type: HEAL_ITEM_BUCKET.to_string(),
+        bucket: bucket.to_string(),
+        disk_count: disks.len(),
+        set_count: 0,
+        ..Default::default()
+    };
+
+    if opts.dry_run {
+        return Ok(res);
+    }
+
+    for (disk, state) in disks.iter().zip(before_state.read().await.iter()) {
+        res.before.drives.push(HealDriveInfo {
+            uuid: "".to_string(),
+            endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(),
+            state: state.to_string(),
+        });
+    }
+
+    if opts.remove && !bucket.starts_with(disk::RUSTFS_META_BUCKET) && !is_all_buckets_not_found(&errs) {
+        let mut futures = Vec::new();
+        for disk in disks.iter() {
+            let disk = disk.clone();
+            let bucket = bucket.to_string();
+            info!("heal_bucket_local, errs: {:?}, opts: {:?}", errs, opts);
+            futures.push(async move {
+                match disk {
+                    Some(disk) => {
+                        info!("will call delete_volume, volume: {}", bucket);
+                        let _ = disk.delete_volume(&bucket).await;
+                        None
+                    }
+                    None => Some(Error::DiskNotFound),
+                }
+            });
+        }
+
+        let _ = join_all(futures).await;
+    }
+
+    if !opts.remove {
+        let mut futures = Vec::new();
+        for (idx, disk) in disks.iter().enumerate() {
+            let disk = disk.clone();
+            let bucket = bucket.to_string();
+            let bs_clone = before_state.clone();
+            let as_clone = after_state.clone();
+            let errs_clone = errs.to_vec();
+            futures.push(async move {
+                if bs_clone.read().await[idx] == DRIVE_STATE_MISSING {
+                    info!("bucket not find, will recreate");
+                    match disk.as_ref().unwrap().make_volume(&bucket).await {
+                        Ok(_) => {
+                            as_clone.write().await[idx] = DRIVE_STATE_OK.to_string();
+                            return None;
+                        }
+                        Err(err) => {
+                            return Some(err);
+                        }
+                    }
+                }
+                errs_clone[idx].clone()
+            });
+        }
+
+        let _ = join_all(futures).await;
+    }
+
+    for (disk, state) in disks.iter().zip(after_state.read().await.iter()) {
+        res.before.drives.push(HealDriveInfo {
+            uuid: "".to_string(),
+            endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(),
+            state: state.to_string(),
+        });
+    }
+
+    Ok(res)
+}
+
+async fn clone_drives() -> Vec<Option<DiskStore>> {
+    GLOBAL_LOCAL_DISK_MAP.read().await.values().cloned().collect::<Vec<_>>()
+}