diff --git a/crates/ecstore/src/disk/endpoint.rs b/crates/ecstore/src/disk/endpoint.rs index 6a8e86e3e..1471c5618 100644 --- a/crates/ecstore/src/disk/endpoint.rs +++ b/crates/ecstore/src/disk/endpoint.rs @@ -1,634 +1 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::disk::error::{Error, Result}; -use path_absolutize::Absolutize; -use rustfs_utils::{is_local_host, is_socket_addr}; -use std::{fmt::Display, path::Path}; -use tracing::debug; -use url::{ParseError, Url}; - -#[cfg(windows)] -pub(crate) fn windows_fallback_local_path( - path: &str, - canonicalize_error: &std::io::Error, - context: &'static str, -) -> std::io::Result { - let absolute = Path::new(path).absolutize()?.to_path_buf(); - tracing::warn!( - path = %path, - canonicalize_error = %canonicalize_error, - resolved = ?absolute, - context = context, - "using windows fallback path resolution for local endpoint" - ); - Ok(absolute) -} - -/// enum for endpoint type. -#[derive(PartialEq, Eq, Debug)] -pub enum EndpointType { - /// path style endpoint type enum. - Path, - - /// URL style endpoint type enum. - Url, -} - -/// any type of endpoint. -#[derive(Debug, PartialEq, Eq, Clone, Hash)] -pub struct Endpoint { - pub url: Url, - pub is_local: bool, - - pub pool_idx: i32, - pub set_idx: i32, - pub disk_idx: i32, -} - -impl Display for Endpoint { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - if self.url.scheme() == "file" { - write!(f, "{}", self.get_file_path()) - } else { - write!(f, "{}", self.url) - } - } -} - -impl TryFrom<&str> for Endpoint { - /// The type returned in the event of a conversion error. - type Error = Error; - - /// Performs the conversion. - fn try_from(value: &str) -> std::result::Result { - // check whether given path is not empty. - if ["", "/", "\\"].iter().any(|&v| v.eq(value)) { - return Err(Error::other("empty or root endpoint is not supported")); - } - - let mut is_local = false; - let url = match Url::parse(value) { - #[allow(unused_mut)] - Ok(mut url) if url.has_host() => { - // URL style of endpoint. - // Valid URL style endpoint is - // - Scheme field must contain "http" or "https" - // - All field should be empty except Host and Path. - if !((url.scheme() == "http" || url.scheme() == "https") - && url.username().is_empty() - && url.fragment().is_none() - && url.query().is_none()) - { - return Err(Error::other("invalid URL endpoint format")); - } - - let path = url.path().to_string(); - - #[cfg(not(windows))] - let path = Path::new(&path).absolutize()?; - - #[cfg(windows)] - let path = if has_leading_slash_windows_drive(&path) { - // Url::path() exposes file-like Windows paths as `/C:/...`. - // Strip only that synthetic leading slash; plain URL paths - // such as `/export1` must stay URL paths, not become - // relative paths under the current drive. - Path::new(&path[1..]).absolutize()?.to_string_lossy().into_owned() - } else { - path - }; - #[cfg(windows)] - let path = Path::new(&path); - - debug!("endpoint try_from: path={}", path.display()); - - if path.parent().is_none() || path.as_os_str().is_empty() { - return Err(Error::other("empty or root path is not supported in URL endpoint")); - } - - match path.to_str() { - Some(v) => url.set_path(v), - None => return Err(Error::other("invalid path")), - } - - url - } - Ok(_) => { - // like d:/foo - is_local = true; - url_parse_from_file_path(value)? - } - Err(e) => match e { - ParseError::InvalidPort => { - return Err(Error::other("invalid URL endpoint format: port number must be between 1 to 65535")); - } - ParseError::EmptyHost => return Err(Error::other("invalid URL endpoint format: empty host name")), - ParseError::RelativeUrlWithoutBase => { - // like /foo - is_local = true; - url_parse_from_file_path(value)? - } - _ => return Err(Error::other(format!("invalid URL endpoint format: {e}"))), - }, - }; - - Ok(Endpoint { - url, - is_local, - pool_idx: -1, - set_idx: -1, - disk_idx: -1, - }) - } -} - -impl Endpoint { - /// returns type of endpoint. - pub fn get_type(&self) -> EndpointType { - if self.url.scheme() == "file" { - EndpointType::Path - } else { - EndpointType::Url - } - } - - /// sets a specific pool number to this node - pub fn set_pool_index(&mut self, idx: usize) { - self.pool_idx = idx as i32 - } - - /// sets a specific set number to this node - pub fn set_set_index(&mut self, idx: usize) { - self.set_idx = idx as i32 - } - - /// sets a specific disk number to this node - pub fn set_disk_index(&mut self, idx: usize) { - self.disk_idx = idx as i32 - } - - /// resolves the host and updates if it is local or not. - pub fn update_is_local(&mut self, local_port: u16) -> Result<()> { - match (self.url.scheme(), self.url.host()) { - (v, Some(host)) if v != "file" => { - self.is_local = is_local_host(host, self.url.port().unwrap_or_default(), local_port)?; - } - _ => {} - } - - Ok(()) - } - - /// returns the host to be used for grid connections. - pub fn grid_host(&self) -> String { - match (self.url.host(), self.url.port()) { - (Some(host), Some(port)) => { - debug!("grid_host scheme={}: host={}, port={}", self.url.scheme(), host, port); - format!("{}://{}:{}", self.url.scheme(), host, port) - } - (Some(host), None) => { - debug!("grid_host scheme={}: host={}", self.url.scheme(), host); - format!("{}://{}", self.url.scheme(), host) - } - _ => String::new(), - } - } - - pub fn host_port(&self) -> String { - match (self.url.host(), self.url.port()) { - (Some(host), Some(port)) => { - debug!("host_port host={}, port={}", host, port); - format!("{host}:{port}") - } - (Some(host), None) => { - debug!("host_port host={}, port={}", host, self.url.port().unwrap_or(0)); - format!("{host}") - } - _ => String::new(), - } - } - - pub fn get_file_path(&self) -> String { - let path: &str = self.url.path(); - let decoded: std::borrow::Cow<'_, str> = match urlencoding::decode(path) { - Ok(decoded) => decoded, - Err(e) => { - debug!("Failed to decode path '{}': {}, using original path", path, e); - std::borrow::Cow::Borrowed(path) - } - }; - #[cfg(windows)] - if self.url.scheme() == "file" { - let stripped: &str = decoded.strip_prefix('/').unwrap_or(&decoded); - debug!("get_file_path windows: path={}", stripped); - return stripped.to_string(); - } - decoded.into_owned() - } -} - -#[cfg(windows)] -fn has_leading_slash_windows_drive(path: &str) -> bool { - let bytes = path.as_bytes(); - bytes.len() >= 4 && bytes[0] == b'/' && bytes[1].is_ascii_alphabetic() && bytes[2] == b':' && bytes[3] == b'/' -} - -/// parse a file path into a URL. -fn url_parse_from_file_path(value: &str) -> Result { - // Only check if the arg is an ip address and ask for scheme since its absent. - // localhost, example.com, any FQDN cannot be disambiguated from a regular file path such as - // /mnt/export1. So we go ahead and start the rustfs server in FS modes in these cases. - let addr: Vec<&str> = value.splitn(2, '/').collect(); - if is_socket_addr(addr[0]) { - return Err(Error::other("invalid URL endpoint format: missing scheme http or https")); - } - - let file_path = match Path::new(value).absolutize() { - Ok(path) => path, - Err(err) => return Err(Error::other(format!("absolute path failed: {err}"))), - }; - - match Url::from_file_path(file_path) { - Ok(url) => Ok(url), - Err(_) => Err(Error::other("Convert a file path into an URL failed")), - } -} - -#[cfg(test)] -mod test { - use super::*; - - fn expected_file_path(path: &str) -> String { - Path::new(path).absolutize().unwrap().to_string_lossy().replace('\\', "/") - } - - fn expected_file_url(path: &str) -> Url { - url_parse_from_file_path(path).unwrap() - } - - #[test] - fn test_new_endpoint() { - #[derive(Default)] - struct TestCase<'a> { - arg: &'a str, - expected_endpoint: Option, - expected_type: Option, - expected_err: Option, - } - - let u2 = Url::parse("https://example.org/path").unwrap(); - let u4 = Url::parse("http://192.168.253.200/path").unwrap(); - let u6 = Url::parse("http://server:/path").unwrap(); - let root_slash_foo = expected_file_url("/foo"); - - let test_cases = [ - TestCase { - arg: "/foo", - expected_endpoint: Some(Endpoint { - url: root_slash_foo, - is_local: true, - pool_idx: -1, - set_idx: -1, - disk_idx: -1, - }), - expected_type: Some(EndpointType::Path), - expected_err: None, - }, - TestCase { - arg: "https://example.org/path", - expected_endpoint: Some(Endpoint { - url: u2, - is_local: false, - pool_idx: -1, - set_idx: -1, - disk_idx: -1, - }), - expected_type: Some(EndpointType::Url), - expected_err: None, - }, - TestCase { - arg: "http://192.168.253.200/path", - expected_endpoint: Some(Endpoint { - url: u4, - is_local: false, - pool_idx: -1, - set_idx: -1, - disk_idx: -1, - }), - expected_type: Some(EndpointType::Url), - expected_err: None, - }, - TestCase { - arg: "", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("empty or root endpoint is not supported")), - }, - TestCase { - arg: "/", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("empty or root endpoint is not supported")), - }, - TestCase { - arg: "\\", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("empty or root endpoint is not supported")), - }, - TestCase { - arg: "c://foo", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format")), - }, - TestCase { - arg: "ftp://foo", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format")), - }, - TestCase { - arg: "http://server/path?location", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format")), - }, - TestCase { - arg: "http://:/path", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format: empty host name")), - }, - TestCase { - arg: "http://:8080/path", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format: empty host name")), - }, - TestCase { - arg: "http://server:/path", - expected_endpoint: Some(Endpoint { - url: u6, - is_local: false, - pool_idx: -1, - set_idx: -1, - disk_idx: -1, - }), - expected_type: Some(EndpointType::Url), - expected_err: None, - }, - TestCase { - arg: "https://93.184.216.34:808080/path", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format: port number must be between 1 to 65535")), - }, - TestCase { - arg: "http://server:8080//", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("empty or root path is not supported in URL endpoint")), - }, - TestCase { - arg: "http://server:8080/", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("empty or root path is not supported in URL endpoint")), - }, - TestCase { - arg: "192.168.1.210:9000", - expected_endpoint: None, - expected_type: None, - expected_err: Some(Error::other("invalid URL endpoint format: missing scheme http or https")), - }, - ]; - - for test_case in test_cases { - let ret = Endpoint::try_from(test_case.arg); - if test_case.expected_err.is_none() && ret.is_err() { - panic!("{}: error: expected = , got = {:?}", test_case.arg, ret); - } - if test_case.expected_err.is_some() && ret.is_ok() { - panic!("{}: error: expected = {:?}, got = ", test_case.arg, test_case.expected_err); - } - match (test_case.expected_err, ret) { - (None, Err(e)) => panic!("{}: error: expected = , got = {}", test_case.arg, e), - (None, Ok(mut ep)) => { - let _ = ep.update_is_local(9000); - if test_case.expected_type != Some(ep.get_type()) { - panic!( - "{}: type: expected = {:?}, got = {:?}", - test_case.arg, - test_case.expected_type, - ep.get_type() - ); - } - - assert_eq!(test_case.expected_endpoint, Some(ep), "{}: endpoint", test_case.arg); - } - (Some(e), Ok(_)) => panic!("{}: error: expected = {}, got = ", test_case.arg, e), - (Some(e), Err(e2)) => { - assert_eq!(e.to_string(), e2.to_string(), "{}: error: expected = {}, got = {}", test_case.arg, e, e2) - } - } - } - } - - #[test] - fn test_endpoint_display() { - // Test file path display - let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); - let display_str = format!("{file_endpoint}"); - assert_eq!(display_str, expected_file_path("/tmp/data")); - - // Test URL display - let url_endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); - let display_str = format!("{url_endpoint}"); - assert_eq!(display_str, "http://example.com:9000/path"); - } - - #[test] - fn test_endpoint_type() { - let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); - assert_eq!(file_endpoint.get_type(), EndpointType::Path); - - let url_endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); - assert_eq!(url_endpoint.get_type(), EndpointType::Url); - } - - #[test] - fn test_endpoint_indexes() { - let mut endpoint = Endpoint::try_from("/tmp/data").unwrap(); - - // Test initial values - assert_eq!(endpoint.pool_idx, -1); - assert_eq!(endpoint.set_idx, -1); - assert_eq!(endpoint.disk_idx, -1); - - // Test setting indexes - endpoint.set_pool_index(2); - endpoint.set_set_index(3); - endpoint.set_disk_index(4); - - assert_eq!(endpoint.pool_idx, 2); - assert_eq!(endpoint.set_idx, 3); - assert_eq!(endpoint.disk_idx, 4); - } - - #[test] - fn test_endpoint_grid_host() { - let endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); - assert_eq!(endpoint.grid_host(), "http://example.com:9000"); - - let endpoint_no_port = Endpoint::try_from("https://example.com/path").unwrap(); - assert_eq!(endpoint_no_port.grid_host(), "https://example.com"); - - let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); - assert_eq!(file_endpoint.grid_host(), ""); - } - - #[test] - fn test_endpoint_host_port() { - let endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); - assert_eq!(endpoint.host_port(), "example.com:9000"); - - let endpoint_no_port = Endpoint::try_from("https://example.com/path").unwrap(); - assert_eq!(endpoint_no_port.host_port(), "example.com"); - - let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); - assert_eq!(file_endpoint.host_port(), ""); - } - - #[test] - fn test_endpoint_get_file_path() { - let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); - assert_eq!(file_endpoint.get_file_path(), expected_file_path("/tmp/data")); - - let url_endpoint = Endpoint::try_from("http://example.com:9000/path/to/data").unwrap(); - assert_eq!(url_endpoint.get_file_path(), "/path/to/data"); - } - - #[cfg(windows)] - #[test] - fn test_windows_url_drive_path_requires_separator_after_colon() { - let drive_path_endpoint = Endpoint::try_from("http://host/C:/data").unwrap(); - assert_eq!(drive_path_endpoint.get_type(), EndpointType::Url); - assert!(has_leading_slash_windows_drive(Url::parse("http://host/C:/data").unwrap().path())); - - let url_path_endpoint = Endpoint::try_from("http://host/C:foo").unwrap(); - assert_eq!(url_path_endpoint.get_type(), EndpointType::Url); - assert!(!has_leading_slash_windows_drive(Url::parse("http://host/C:foo").unwrap().path())); - assert_eq!(url_path_endpoint.get_file_path(), "/C:foo"); - } - - #[test] - fn test_endpoint_clone_and_equality() { - let endpoint1 = Endpoint::try_from("/tmp/data").unwrap(); - let endpoint2 = endpoint1.clone(); - - assert_eq!(endpoint1, endpoint2); - assert_eq!(endpoint1.url, endpoint2.url); - assert_eq!(endpoint1.is_local, endpoint2.is_local); - assert_eq!(endpoint1.pool_idx, endpoint2.pool_idx); - assert_eq!(endpoint1.set_idx, endpoint2.set_idx); - assert_eq!(endpoint1.disk_idx, endpoint2.disk_idx); - } - - #[test] - fn test_endpoint_with_special_paths() { - // Test with complex paths - let complex_path = "/var/lib/rustfs/data/bucket1"; - let endpoint = Endpoint::try_from(complex_path).unwrap(); - assert_eq!(endpoint.get_file_path(), expected_file_path(complex_path)); - assert!(endpoint.is_local); - assert_eq!(endpoint.get_type(), EndpointType::Path); - } - - #[test] - fn test_endpoint_with_spaces_in_path() { - let path_with_spaces = "/Users/test/Library/Application Support/rustfs/data"; - let endpoint = Endpoint::try_from(path_with_spaces).unwrap(); - assert_eq!(endpoint.get_file_path(), expected_file_path(path_with_spaces)); - assert!(endpoint.is_local); - assert_eq!(endpoint.get_type(), EndpointType::Path); - } - - #[test] - fn test_endpoint_percent_encoding_roundtrip() { - let path_with_spaces = "/Users/test/Library/Application Support/rustfs/data"; - let endpoint = Endpoint::try_from(path_with_spaces).unwrap(); - - // Verify that the URL internally stores percent-encoded path - assert!( - endpoint.url.path().contains("%20"), - "URL path should contain percent-encoded spaces: {}", - endpoint.url.path() - ); - - // Verify that get_file_path() decodes the percent-encoded path correctly - assert_eq!( - endpoint.get_file_path(), - expected_file_path("/Users/test/Library/Application Support/rustfs/data"), - "get_file_path() should decode percent-encoded spaces" - ); - } - - #[test] - fn test_endpoint_with_various_special_characters() { - // Test path with multiple special characters that get percent-encoded - let path_with_special = "/tmp/test path/data[1]/file+name&more"; - let endpoint = Endpoint::try_from(path_with_special).unwrap(); - - // get_file_path() should return the original path with decoded characters - assert_eq!(endpoint.get_file_path(), expected_file_path(path_with_special)); - } - - #[test] - fn test_endpoint_update_is_local() { - let mut endpoint = Endpoint::try_from("http://localhost:9000/path").unwrap(); - let result = endpoint.update_is_local(9000); - assert!(result.is_ok()); - - let mut file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); - let result = file_endpoint.update_is_local(9000); - assert!(result.is_ok()); - } - - #[test] - fn test_url_parse_from_file_path() { - let result = url_parse_from_file_path("/tmp/test"); - assert!(result.is_ok()); - - let url = result.unwrap(); - assert_eq!(url.scheme(), "file"); - } - - #[test] - fn test_endpoint_hash() { - use std::collections::HashSet; - - let endpoint1 = Endpoint::try_from("/tmp/data1").unwrap(); - let endpoint2 = Endpoint::try_from("/tmp/data2").unwrap(); - let endpoint3 = endpoint1.clone(); - - let mut set = HashSet::new(); - set.insert(endpoint1); - set.insert(endpoint2); - set.insert(endpoint3); // Should not be added as it's equal to endpoint1 - - assert_eq!(set.len(), 2); - } -} +pub use crate::layout::endpoint::*; diff --git a/crates/ecstore/src/endpoints.rs b/crates/ecstore/src/endpoints.rs index 1a39f21fe..a29173960 100644 --- a/crates/ecstore/src/endpoints.rs +++ b/crates/ecstore/src/endpoints.rs @@ -1,1696 +1 @@ -// Copyright 2024 RustFS Team -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -use crate::{ - disk::endpoint::{Endpoint, EndpointType}, - disks_layout::DisksLayout, - global::global_rustfs_port, -}; -use rustfs_config::{DEFAULT_UNSAFE_BYPASS_DISK_CHECK, ENV_MINIO_CI, ENV_UNSAFE_BYPASS_DISK_CHECK}; -use rustfs_utils::{XHost, check_local_server_addr, get_host_ip, is_local_host}; -use std::{ - collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry}, - io::{Error, ErrorKind, Result}, - net::IpAddr, -}; -use tracing::{error, info, instrument, warn}; - -/// enum for setup type. -#[derive(PartialEq, Eq, Debug, Clone)] -pub enum SetupType { - /// starts with unknown setup type. - Unknown, - - /// FS setup type enum. - FS, - - /// Erasure single drive setup enum. - ErasureSD, - - /// Erasure setup type enum. - Erasure, - - /// Distributed Erasure setup type enum. - DistErasure, -} - -/// holds information about a node in this cluster -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct Node { - pub url: url::Url, - pub pools: Vec, - pub is_local: bool, - pub grid_host: String, -} - -/// list of same type of endpoint. -#[derive(Debug, Default, Clone)] -pub struct Endpoints(Vec); - -#[derive(Debug, Clone)] -struct LocalDiskValidationDiagnostic { - original_path: String, - canonical_path: Option, - device_numbers: Option, - device_ids: Option>, -} - -impl LocalDiskValidationDiagnostic { - fn new(original_path: &str) -> Self { - Self { - original_path: original_path.to_string(), - canonical_path: None, - device_numbers: None, - device_ids: None, - } - } - - fn summary(&self) -> String { - let canonical_path = self.canonical_path.as_deref().unwrap_or("(unresolved)"); - let device_numbers = self.device_numbers.as_deref().unwrap_or("(unavailable)"); - let device_ids = self - .device_ids - .as_ref() - .map(|ids| ids.join(",")) - .unwrap_or_else(|| "(unavailable)".to_string()); - - format!( - "path='{}', canonical='{}', st_dev='{}', device_ids=[{}]", - self.original_path, canonical_path, device_numbers, device_ids - ) - } -} - -impl AsRef> for Endpoints { - fn as_ref(&self) -> &Vec { - &self.0 - } -} - -impl AsMut> for Endpoints { - fn as_mut(&mut self) -> &mut Vec { - &mut self.0 - } -} - -impl From> for Endpoints { - fn from(v: Vec) -> Self { - Self(v) - } -} - -impl> TryFrom<&[T]> for Endpoints { - type Error = Error; - - /// returns new endpoint list based on input args. - fn try_from(args: &[T]) -> Result { - let mut endpoint_type = None; - let mut schema = None; - let mut endpoints = Vec::with_capacity(args.len()); - let mut uniq_set = HashSet::with_capacity(args.len()); - - // Loop through args and adds to endpoint list. - for (i, arg) in args.iter().enumerate() { - let endpoint = match Endpoint::try_from(arg.as_ref()) { - Ok(ep) => ep, - Err(e) => return Err(Error::other(format!("'{}': {}", arg.as_ref(), e))), - }; - - // All endpoints have to be same type and scheme if applicable. - if i == 0 { - endpoint_type = Some(endpoint.get_type()); - schema = Some(endpoint.url.scheme().to_owned()); - } else if Some(endpoint.get_type()) != endpoint_type { - return Err(Error::other("mixed style endpoints are not supported")); - } else if Some(endpoint.url.scheme()) != schema.as_deref() { - return Err(Error::other("mixed scheme is not supported")); - } - - // Check for duplicate endpoints. - let endpoint_str = endpoint.to_string(); - if uniq_set.contains(&endpoint_str) { - return Err(Error::other("duplicate endpoints found")); - } - - uniq_set.insert(endpoint_str); - endpoints.push(endpoint); - } - - Ok(Endpoints(endpoints)) - } -} - -impl Endpoints { - /// Converts `self` into its inner representation. - /// - /// This method consumes the `self` object and returns its inner `Vec`. - /// It is useful for when you need to take the endpoints out of their container - /// without needing a reference to the container itself. - pub fn into_inner(self) -> Vec { - self.0 - } - - pub fn into_ref(&self) -> &Vec { - &self.0 - } - - // GetString - returns endpoint string of i-th endpoint (0-based), - // and empty string for invalid indexes. - pub fn get_string(&self, i: usize) -> String { - if i >= self.0.len() { - return "".to_string(); - } - - self.0[i].to_string() - } - - pub fn is_empty(&self) -> bool { - self.0.is_empty() - } -} - -#[derive(Debug)] -/// a temporary type to holds the list of endpoints -struct PoolEndpointList { - inner: Vec, - setup_type: SetupType, -} - -impl AsRef> for PoolEndpointList { - fn as_ref(&self) -> &Vec { - &self.inner - } -} - -impl AsMut> for PoolEndpointList { - fn as_mut(&mut self) -> &mut Vec { - &mut self.inner - } -} - -impl PoolEndpointList { - /// creates a list of endpoints per pool, resolves their relevant - /// hostnames and discovers those are local or remote. - async fn create_pool_endpoints(server_addr: &str, disks_layout: &DisksLayout) -> Result { - if disks_layout.is_empty_layout() { - return Err(Error::other("invalid number of endpoints")); - } - - let server_addr = check_local_server_addr(server_addr)?; - - // For single arg, return single drive EC setup. - if disks_layout.is_single_drive_layout() { - let mut endpoint = Endpoint::try_from(disks_layout.get_single_drive_layout())?; - endpoint.update_is_local(server_addr.port())?; - - if endpoint.get_type() != EndpointType::Path { - return Err(Error::other("use path style endpoint for single node setup")); - } - - endpoint.set_pool_index(0); - endpoint.set_set_index(0); - endpoint.set_disk_index(0); - - // TODO Check for cross device mounts if any. - - return Ok(Self { - inner: vec![Endpoints::from(vec![endpoint])], - setup_type: SetupType::ErasureSD, - }); - } - - let mut pool_endpoints = Vec::::with_capacity(disks_layout.pools.len()); - for (pool_idx, pool) in disks_layout.pools.iter().enumerate() { - let mut endpoints = Endpoints::default(); - for (set_idx, set_layout) in pool.iter().enumerate() { - // Convert args to endpoints - let mut eps = Endpoints::try_from(set_layout.as_slice())?; - - // TODO Check for cross device mounts if any. - - for (disk_idx, ep) in eps.as_mut().iter_mut().enumerate() { - ep.set_pool_index(pool_idx); - ep.set_set_index(set_idx); - ep.set_disk_index(disk_idx); - } - - endpoints.as_mut().append(eps.as_mut()); - } - - if endpoints.as_ref().is_empty() { - return Err(Error::other("invalid number of endpoints")); - } - - pool_endpoints.push(endpoints); - } - - // setup type - let mut unique_args = HashSet::new(); - let mut pool_endpoint_list = Self { - inner: pool_endpoints, - setup_type: SetupType::Unknown, - }; - - pool_endpoint_list.update_is_local(server_addr.port())?; - - for endpoints in pool_endpoint_list.inner.iter_mut() { - // Check whether same path is not used in endpoints of a host on different port. - let mut path_ip_map: HashMap> = HashMap::new(); - let mut host_ip_cache = HashMap::new(); - for ep in endpoints.as_ref() { - if !ep.url.has_host() { - continue; - } - - let host = ep.url.host().unwrap(); - let host_ip_set = if let Some(set) = host_ip_cache.get(&host) { - info!( - target: "rustfs::ecstore::endpoints", - host = %host, - endpoint = %ep.to_string(), - from = "cache", - "Create pool endpoints host '{}' found in cache for endpoint '{}'", host, ep.to_string() - ); - set - } else { - let ips = match get_host_ip(host.clone()).await { - Ok(ips) => ips, - Err(e) => { - error!("Create pool endpoints host {} not found, error:{}", host, e); - return Err(Error::other(format!("host '{host}' cannot resolve: {e}"))); - } - }; - info!( - target: "rustfs::ecstore::endpoints", - host = %host, - endpoint = %ep.to_string(), - from = "get_host_ip", - "Create pool endpoints host '{}' resolved to ips {:?} for endpoint '{}'", - host, - ips, - ep.to_string() - ); - host_ip_cache.insert(host.clone(), ips); - host_ip_cache.get(&host).unwrap() - }; - - let path = ep.get_file_path(); - match path_ip_map.entry(path) { - Entry::Occupied(mut e) => { - if e.get().intersection(host_ip_set).count() > 0 { - let path_key = e.key().clone(); - return Err(Error::other(format!( - "same path '{path_key}' can not be served by different port on same address" - ))); - } - e.get_mut().extend(host_ip_set.iter()); - } - Entry::Vacant(e) => { - e.insert(host_ip_set.clone()); - } - } - } - - // Check whether same path is used for more than 1 local endpoints. - let mut local_path_set = HashSet::new(); - for ep in endpoints.as_ref() { - if !ep.is_local { - continue; - } - - let path = ep.get_file_path(); - if local_path_set.contains(&path) { - return Err(Error::other(format!( - "path '{path}' cannot be served by different address on same server" - ))); - } - local_path_set.insert(path); - } - - // Here all endpoints are URL style. - let mut ep_path_set = HashSet::new(); - let mut local_server_host_set = HashSet::new(); - let mut local_port_set = HashSet::new(); - let mut local_endpoint_count = 0; - - for ep in endpoints.as_ref() { - ep_path_set.insert(ep.get_file_path()); - if ep.is_local && ep.url.has_host() { - local_server_host_set.insert(ep.url.host()); - local_port_set.insert(ep.url.port()); - local_endpoint_count += 1; - } - } - - // All endpoints are pointing to local host - if endpoints.as_ref().len() == local_endpoint_count { - // If all endpoints have same port number, Just treat it as local erasure setup - // using URL style endpoints. - if local_port_set.len() == 1 && local_server_host_set.len() > 1 { - return Err(Error::other("all local endpoints should not have different hostnames/ips")); - } - } - - // Add missing port in all endpoints. - for ep in endpoints.as_mut() { - if !ep.url.has_host() { - unique_args.insert(format!("localhost:{}", server_addr.port())); - continue; - } - match ep.url.port() { - None => { - let _ = ep.url.set_port(Some(server_addr.port())); - } - Some(port) => { - // If endpoint is local, but port is different than serverAddrPort, then make it as remote. - if ep.is_local && server_addr.port() != port { - ep.is_local = false; - } - } - } - unique_args.insert(ep.host_port()); - } - } - - validate_local_physical_disk_independence(pool_endpoint_list.as_ref())?; - - let setup_type = match pool_endpoint_list.as_ref()[0].as_ref()[0].get_type() { - EndpointType::Path => SetupType::Erasure, - EndpointType::Url => match unique_args.len() { - 1 => SetupType::Erasure, - _ => SetupType::DistErasure, - }, - }; - - pool_endpoint_list.setup_type = setup_type; - - Ok(pool_endpoint_list) - } - - /// resolves all hosts and discovers which are local - fn update_is_local(&mut self, local_port: u16) -> Result<()> { - for endpoints in self.inner.iter_mut() { - for ep in endpoints.as_mut() { - match ep.url.host() { - None => { - ep.is_local = true; - } - Some(host) => { - ep.is_local = is_local_host(host, ep.url.port().unwrap_or_default(), local_port)?; - } - } - } - } - - Ok(()) - } - - /// resolves all hosts and discovers which are local - fn _update_is_local(&mut self, local_port: u16) -> Result<()> { - let mut eps_resolved = 0; - let mut found_local = false; - let mut resolved_set: HashSet<(usize, usize)> = HashSet::new(); - let ep_count: usize = self.inner.iter().map(|v| v.as_ref().len()).sum(); - - loop { - // Break if the local endpoint is found already Or all the endpoints are resolved. - if found_local || eps_resolved == ep_count { - break; - } - - for (i, endpoints) in self.inner.iter_mut().enumerate() { - for (j, ep) in endpoints.as_mut().iter_mut().enumerate() { - if resolved_set.contains(&(i, j)) { - // Continue if host is already resolved. - continue; - } - - match ep.url.host() { - None => { - if !found_local { - found_local = true; - } - ep.is_local = true; - eps_resolved += 1; - resolved_set.insert((i, j)); - continue; - } - Some(host) => match is_local_host(host, ep.url.port().unwrap_or_default(), local_port) { - Ok(is_local) => { - if !found_local { - found_local = is_local; - } - ep.is_local = is_local; - eps_resolved += 1; - resolved_set.insert((i, j)); - } - Err(err) => { - // TODO Retry infinitely on Kubernetes and Docker swarm? - return Err(err); - } - }, - } - } - } - } - - Ok(()) - } -} - -/// represent endpoints in a given pool -/// along with its setCount and setDriveCount. -#[derive(Debug, Clone)] -pub struct PoolEndpoints { - // indicates if endpoints are provided in non-ellipses style - pub legacy: bool, - pub set_count: usize, - pub drives_per_set: usize, - pub endpoints: Endpoints, - pub cmd_line: String, - pub platform: String, -} - -/// list of endpoints -#[derive(Debug, Clone, Default)] -pub struct EndpointServerPools(pub Vec); - -impl From> for EndpointServerPools { - fn from(v: Vec) -> Self { - Self(v) - } -} - -impl AsRef> for EndpointServerPools { - fn as_ref(&self) -> &Vec { - &self.0 - } -} - -impl AsMut> for EndpointServerPools { - fn as_mut(&mut self) -> &mut Vec { - &mut self.0 - } -} - -impl EndpointServerPools { - pub fn reset(&mut self, eps: Vec) { - self.0 = eps; - } - pub fn legacy(&self) -> bool { - self.0.len() == 1 && self.0[0].legacy - } - pub fn get_pool_idx(&self, cmd_line: &str) -> Option { - for (idx, eps) in self.0.iter().enumerate() { - if eps.cmd_line.as_str() == cmd_line { - return Some(idx); - } - } - None - } - pub async fn from_volumes(server_addr: &str, endpoints: Vec) -> Result<(EndpointServerPools, SetupType)> { - let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; - - Self::create_server_endpoints(server_addr, &layouts).await - } - /// validates and creates new endpoints from input args, supports - /// both ellipses and without ellipses transparently. - pub async fn create_server_endpoints( - server_addr: &str, - disks_layout: &DisksLayout, - ) -> Result<(EndpointServerPools, SetupType)> { - if disks_layout.pools.is_empty() { - return Err(Error::other("Invalid arguments specified")); - } - - let pool_eps = PoolEndpointList::create_pool_endpoints(server_addr, disks_layout).await?; - - let mut ret: EndpointServerPools = Vec::with_capacity(pool_eps.as_ref().len()).into(); - for (i, eps) in pool_eps.inner.into_iter().enumerate() { - let ep = PoolEndpoints { - legacy: disks_layout.legacy, - set_count: disks_layout.get_set_count(i), - drives_per_set: disks_layout.get_drives_per_set(i), - endpoints: eps, - cmd_line: disks_layout.get_cmd_line(i), - platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), - }; - - ret.add(ep)?; - } - - Ok((ret, pool_eps.setup_type)) - } - - pub fn es_count(&self) -> usize { - self.0.iter().map(|v| v.set_count).sum() - } - - /// add pool endpoints - pub fn add(&mut self, eps: PoolEndpoints) -> Result<()> { - let mut exits = HashSet::new(); - for peps in self.0.iter() { - for ep in peps.endpoints.as_ref() { - exits.insert(ep.to_string()); - } - } - - for ep in eps.endpoints.as_ref() { - if exits.contains(&ep.to_string()) { - return Err(Error::other("duplicate endpoints found")); - } - } - - self.0.push(eps); - - Ok(()) - } - - /// returns true if the first endpoint is local. - pub fn first_local(&self) -> bool { - self.0 - .first() - .and_then(|v| v.endpoints.as_ref().first()) - .is_some_and(|v| v.is_local) - } - - /// returns a sorted list of nodes in this cluster - pub fn get_nodes(&self) -> Vec { - let mut node_map = HashMap::new(); - - for pool in self.0.iter() { - for ep in pool.endpoints.as_ref() { - let n = node_map.entry(ep.host_port()).or_insert_with(|| Node { - url: ep.url.clone(), - pools: vec![], - is_local: ep.is_local, - grid_host: ep.grid_host(), - }); - - if !n.pools.contains(&(ep.pool_idx as usize)) { - n.pools.push(ep.pool_idx as usize); - } - } - } - - let mut nodes: Vec = node_map.into_values().collect(); - - nodes.sort_by(|a, b| a.grid_host.cmp(&b.grid_host)); - - nodes - } - - #[instrument] - pub fn hosts_sorted(&self) -> Vec> { - let (mut peers, local) = self.peers(); - - let mut ret = vec![None; peers.len()]; - - peers.sort(); - - for (i, peer) in peers.iter().enumerate() { - if &local == peer { - continue; - } - - let host = match XHost::try_from(peer.clone()) { - Ok(res) => res, - Err(err) => { - warn!("Xhost parse failed {:?}", err); - continue; - } - }; - - ret[i] = Some(host); - } - - ret - } - pub fn peers(&self) -> (Vec, 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 && endpoint.url.port() == Some(global_rustfs_port()) && local.is_none() { - local = Some(host.clone()); - } - - set.insert(host); - } - } - - let hosts: Vec = set.iter().cloned().collect(); - - (hosts, local.unwrap_or_default()) - } - - pub fn find_grid_hosts_from_peer(&self, host: &XHost) -> Option { - for ep in self.0.iter() { - for endpoint in ep.endpoints.0.iter() { - if endpoint.is_local { - continue; - } - let xhost = match XHost::try_from(endpoint.host_port()) { - Ok(res) => res, - Err(_) => { - continue; - } - }; - - if xhost.to_string() == host.to_string() { - return Some(endpoint.grid_host()); - } - } - } - - None - } -} - -fn validate_local_physical_disk_independence(pools: &[Endpoints]) -> Result<()> { - let mut local_paths = BTreeSet::new(); - for endpoints in pools { - for endpoint in endpoints.as_ref() { - if endpoint.is_local { - local_paths.insert(endpoint.get_file_path()); - } - } - } - - if local_paths.is_empty() { - return Ok(()); - } - - let local_paths = local_paths.into_iter().collect::>(); - validate_local_cross_device_mounts(&local_paths)?; - - if local_paths.len() <= 1 { - return Ok(()); - } - - // Compatibility behavior: - // - canonical key: RUSTFS_UNSAFE_BYPASS_DISK_CHECK - // - legacy CI alias: MINIO_CI - // If both are set, `get_env_bool_with_aliases` keeps canonical key precedence. - if rustfs_utils::get_env_bool_with_aliases(ENV_UNSAFE_BYPASS_DISK_CHECK, &[ENV_MINIO_CI], DEFAULT_UNSAFE_BYPASS_DISK_CHECK) { - warn!( - env = ENV_UNSAFE_BYPASS_DISK_CHECK, - local_paths = ?local_paths, - "Skipping local physical disk independence validation due to explicit environment override", - ); - return Ok(()); - } - - let mut device_paths = BTreeMap::>::new(); - let mut diagnostics = Vec::with_capacity(local_paths.len()); - #[cfg(not(windows))] - let mut missing_paths = Vec::new(); - - for path in &local_paths { - let mut diagnostic = LocalDiskValidationDiagnostic::new(path); - let canonical = match rustfs_utils::canonicalize(path) { - Ok(path) => path, - Err(err) if err.kind() == ErrorKind::NotFound => { - // On Windows, canonicalize can fail for ZFS volumes, junction points, - // subst drives, and other non-standard mounts. Try absolutize as fallback. - #[cfg(windows)] - { - match crate::disk::endpoint::windows_fallback_local_path(path, &err, "disk independence validation") { - Ok(absolute) => { - let abs_path = absolute.to_string_lossy().into_owned(); - diagnostic.canonical_path = Some(abs_path.clone()); - if let Ok(serial) = rustfs_utils::os::get_volume_serial_number(&abs_path) { - diagnostic.device_numbers = Some(format!("serial:{serial:#010x}")); - } - match rustfs_utils::os::get_physical_device_ids(&abs_path) { - Ok(ids) => { - diagnostic.device_ids = Some(ids.clone()); - for device_id in ids { - device_paths.entry(device_id).or_default().insert(abs_path.clone()); - } - } - Err(device_err) => { - return Err(Error::other(format!( - "failed to inspect physical disk for local endpoint '{abs_path}' after fallback path resolution: {device_err}" - ))); - } - } - diagnostics.push(diagnostic); - continue; - } - Err(fallback_err) => { - return Err(Error::other(format!( - "failed to resolve local endpoint path '{path}' for disk validation: {err}; fallback resolution failed: {fallback_err}" - ))); - } - } - } - #[cfg(not(windows))] - { - missing_paths.push(path.clone()); - diagnostics.push(diagnostic); - continue; - } - } - Err(err) => { - return Err(Error::other(format!( - "failed to resolve local endpoint path '{path}' for disk validation: {err}" - ))); - } - }; - let canonical_path = canonical.to_string_lossy().into_owned(); - diagnostic.canonical_path = Some(canonical_path.clone()); - #[cfg(not(windows))] - if let Ok(stat) = rustix::fs::stat(canonical.as_path()) { - diagnostic.device_numbers = Some(format!("{}:{}", rustix::fs::major(stat.st_dev), rustix::fs::minor(stat.st_dev))); - } - #[cfg(windows)] - if let Ok(serial) = rustfs_utils::os::get_volume_serial_number(&canonical_path) { - diagnostic.device_numbers = Some(format!("serial:{serial:#010x}")); - } - let device_ids = rustfs_utils::os::get_physical_device_ids(&canonical_path).map_err(|err| { - Error::other(format!("failed to inspect physical disk for local endpoint '{canonical_path}': {err}")) - })?; - diagnostic.device_ids = Some(device_ids.clone()); - diagnostics.push(diagnostic); - - for device_id in device_ids { - device_paths.entry(device_id).or_default().insert(canonical_path.clone()); - } - } - - #[cfg(not(windows))] - if !missing_paths.is_empty() { - warn!( - missing_paths = ?missing_paths, - "Excluding non-existent local endpoint paths from physical disk independence validation during endpoint parsing", - ); - } - - warn!( - diagnostics = %diagnostics - .iter() - .map(LocalDiskValidationDiagnostic::summary) - .collect::>() - .join("; "), - "Collected local endpoint disk-topology diagnostics before physical disk independence validation", - ); - - let shared_devices = device_paths - .into_iter() - .filter_map(|(device_id, paths)| { - if paths.len() <= 1 { - return None; - } - - Some((device_id, paths.into_iter().collect::>())) - }) - .collect::>(); - - if shared_devices.is_empty() { - return Ok(()); - } - - let details = shared_devices - .into_iter() - .map(|(device_id, paths)| format!("{device_id} => {}", paths.join(", "))) - .collect::>() - .join("; "); - let diagnostics_summary = diagnostics - .iter() - .map(LocalDiskValidationDiagnostic::summary) - .collect::>() - .join("; "); - - Err(Error::other(format!( - "local erasure endpoints must use distinct physical disks; detected shared devices [{details}]. \ -validation diagnostics: [{diagnostics_summary}]. \ -Set {ENV_UNSAFE_BYPASS_DISK_CHECK}=true only for local testing or CI to bypass this safety check" - ))) -} - -fn validate_local_cross_device_mounts(local_paths: &[String]) -> Result<()> { - rustfs_utils::os::check_cross_device_mounts(local_paths) - .map_err(|err| Error::other(format!("local endpoint cross-device mount validation failed: {err}"))) -} - -#[cfg(test)] -mod test { - use path_absolutize::Absolutize; - use rustfs_utils::must_get_local_ips; - - use super::*; - - #[cfg(target_os = "linux")] - use serial_test::serial; - use std::path::Path; - #[cfg(target_os = "linux")] - use temp_env::async_with_vars; - #[cfg(target_os = "linux")] - use tempfile::tempdir; - - #[test] - fn test_new_endpoints() { - let test_cases = [ - (vec!["/d1", "/d2", "/d3", "/d4"], None, 1), - ( - vec![ - "http://localhost/d1", - "http://localhost/d2", - "http://localhost/d3", - "http://localhost/d4", - ], - None, - 2, - ), - ( - vec![ - "http://example.org/d1", - "http://example.com/d1", - "http://example.net/d1", - "http://example.edu/d1", - ], - None, - 3, - ), - ( - vec![ - "http://localhost/d1", - "http://localhost/d2", - "http://example.org/d1", - "http://example.org/d2", - ], - None, - 4, - ), - ( - vec![ - "https://localhost:9000/d1", - "https://localhost:9001/d2", - "https://localhost:9002/d3", - "https://localhost:9003/d4", - ], - None, - 5, - ), - // It is valid WRT endpoint list that same path is expected with different port on same server. - ( - vec![ - "https://127.0.0.1:9000/d1", - "https://127.0.0.1:9001/d1", - "https://127.0.0.1:9002/d1", - "https://127.0.0.1:9003/d1", - ], - None, - 6, - ), - (vec!["d1", "d2", "d3", "d1"], Some(Error::other("duplicate endpoints found")), 7), - (vec!["d1", "d2", "d3", "./d1"], Some(Error::other("duplicate endpoints found")), 8), - ( - vec![ - "http://localhost/d1", - "http://localhost/d2", - "http://localhost/d1", - "http://localhost/d4", - ], - Some(Error::other("duplicate endpoints found")), - 9, - ), - ( - vec!["ftp://server/d1", "http://server/d2", "http://server/d3", "http://server/d4"], - Some(Error::other("'ftp://server/d1': io error invalid URL endpoint format")), - 10, - ), - ( - vec!["d1", "http://localhost/d2", "d3", "d4"], - Some(Error::other("mixed style endpoints are not supported")), - 11, - ), - ( - vec![ - "http://example.org/d1", - "https://example.com/d1", - "http://example.net/d1", - "https://example.edut/d1", - ], - Some(Error::other("mixed scheme is not supported")), - 12, - ), - ( - vec![ - "192.168.1.210:9000/tmp/dir0", - "192.168.1.210:9000/tmp/dir1", - "192.168.1.210:9000/tmp/dir2", - "192.168.110:9000/tmp/dir3", - ], - Some(Error::other("'192.168.1.210:9000/tmp/dir0': io error")), - 13, - ), - ]; - - for test_case in test_cases { - let args: Vec = test_case.0.iter().map(|v| v.to_string()).collect(); - let ret = Endpoints::try_from(args.as_slice()); - - match (test_case.1, ret) { - (None, Err(e)) => panic!("{}: error: expected = , got = {}", test_case.2, e), - (None, Ok(_)) => {} - (Some(e), Ok(_)) => panic!("{}: error: expected = {}, got = ", test_case.2, e), - (Some(e), Err(e2)) => { - assert!( - e2.to_string().starts_with(&e.to_string()), - "{}: error: expected = {}, got = {}", - test_case.2, - e, - e2 - ) - } - } - } - } - - #[tokio::test] - async fn test_create_pool_endpoints() { - #[derive(Default)] - struct TestCase<'a> { - num: usize, - server_addr: &'a str, - args: Vec<&'a str>, - expected_endpoints: Option, - expected_setup_type: Option, - expected_err: Option, - } - - // Filter ipList by IPs those do not start with '127.'. - let non_loop_back_i_ps = - must_get_local_ips().map_or(vec![], |v| v.into_iter().filter(|ip| ip.is_ipv4() && ip.is_loopback()).collect()); - if non_loop_back_i_ps.is_empty() { - panic!("No non-loop back IP address found for this host"); - } - let non_loop_back_ip = non_loop_back_i_ps[0]; - let remote_ip1 = "192.0.2.10"; - let remote_ip2 = "192.0.2.11"; - let remote_ip3 = "192.0.2.12"; - - let case1_endpoint1 = format!("http://{non_loop_back_ip}/d1"); - let case1_endpoint2 = format!("http://{non_loop_back_ip}/d2"); - let args = vec![ - format!("http://{}:10000/d1", non_loop_back_ip), - format!("http://{}:10000/d2", non_loop_back_ip), - format!("http://{remote_ip1}:10000/d3"), - format!("http://{remote_ip2}:10000/d4"), - ]; - let (case1_ur_ls, case1_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:10000/")); - - let case2_endpoint1 = format!("http://{non_loop_back_ip}/d1"); - let case2_endpoint2 = format!("http://{non_loop_back_ip}:9000/d2"); - let args = vec![ - format!("http://{}:10000/d1", non_loop_back_ip), - format!("http://{}:9000/d2", non_loop_back_ip), - format!("http://{remote_ip1}:10000/d3"), - format!("http://{remote_ip2}:10000/d4"), - ]; - let (case2_ur_ls, case2_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:10000/")); - - let case3_endpoint1 = format!("http://{non_loop_back_ip}/d1"); - let args = vec![ - format!("http://{}:80/d1", non_loop_back_ip), - format!("http://{remote_ip1}:9000/d2"), - format!("http://{remote_ip2}:80/d3"), - format!("http://{remote_ip3}:80/d4"), - ]; - let (case3_ur_ls, case3_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:80/")); - - let case4_endpoint1 = format!("http://{non_loop_back_ip}/d1"); - let args = vec![ - format!("http://{}:9000/d1", non_loop_back_ip), - format!("http://{remote_ip1}:9000/d2"), - format!("http://{remote_ip2}:9000/d3"), - format!("http://{remote_ip3}:9000/d4"), - ]; - let (case4_ur_ls, case4_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9000/")); - - let case5_endpoint1 = format!("http://{non_loop_back_ip}:9000/d1"); - let case5_endpoint2 = format!("http://{non_loop_back_ip}:9001/d2"); - let case5_endpoint3 = format!("http://{non_loop_back_ip}:9002/d3"); - let case5_endpoint4 = format!("http://{non_loop_back_ip}:9003/d4"); - let args = vec![ - case5_endpoint1.clone(), - case5_endpoint2.clone(), - case5_endpoint3.clone(), - case5_endpoint4.clone(), - ]; - let (case5_ur_ls, case5_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9000/")); - - let case6_endpoint1 = format!("http://{non_loop_back_ip}:9003/d4"); - let args = vec![ - "http://127.0.0.1:9000/d1".to_string(), - "http://127.0.0.1:9001/d2".to_string(), - "http://127.0.0.1:9002/d3".to_string(), - case6_endpoint1.clone(), - ]; - let (case6_ur_ls, case6_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9003/")); - - let case7_endpoint1 = format!("http://{non_loop_back_ip}:9001/export"); - let case7_endpoint2 = format!("http://{non_loop_back_ip}:9000/export"); - - let test_cases = [ - TestCase { - num: 1, - server_addr: "localhost", - expected_err: Some(Error::other("address localhost: missing port in address")), - ..Default::default() - }, - // Erasure Single Drive - TestCase { - num: 2, - server_addr: "127.0.0.1:9000", - args: vec!["http://127.0.0.1/d1"], - expected_err: Some(Error::other("use path style endpoint for single node setup")), - ..Default::default() - }, - TestCase { - num: 3, - server_addr: "0.0.0.0:443", - args: vec!["/d1"], - expected_endpoints: Some(Endpoints(vec![Endpoint { - url: must_file_path("/d1"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }])), - expected_setup_type: Some(SetupType::ErasureSD), - ..Default::default() - }, - TestCase { - num: 4, - server_addr: "127.0.0.1:10000", - args: vec!["/d1"], - expected_endpoints: Some(Endpoints(vec![Endpoint { - url: must_file_path("/d1"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }])), - expected_setup_type: Some(SetupType::ErasureSD), - ..Default::default() - }, - TestCase { - num: 5, - server_addr: "127.0.0.1:9000", - args: vec![ - "https://127.0.0.1:9000/d1", - "https://127.0.0.1:9001/d1", - "https://192.0.2.1/d1", - "https://192.0.2.1/d2", - ], - expected_err: Some(Error::other("same path '/d1' can not be served by different port on same address")), - ..Default::default() - }, - // Erasure Setup with PathEndpointType - TestCase { - num: 6, - server_addr: "0.0.0.0:1234", - args: vec!["/d1", "/d2", "/d3", "/d4"], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: must_file_path("/d1"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: must_file_path("/d2"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: must_file_path("/d3"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: must_file_path("/d4"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::Erasure), - ..Default::default() - }, - // DistErasure Setup with URLEndpointType - TestCase { - num: 7, - server_addr: "0.0.0.0:9000", - args: vec![ - "http://127.0.0.1/d1", - "http://127.0.0.1/d2", - "http://127.0.0.1/d3", - "http://127.0.0.1/d4", - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: must_url("http://127.0.0.1:9000/d1"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: must_url("http://127.0.0.1:9000/d2"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: must_url("http://127.0.0.1:9000/d3"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: must_url("http://127.0.0.1:9000/d4"), - is_local: true, - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::Erasure), - ..Default::default() - }, - // DistErasure Setup with URLEndpointType having mixed naming to local host. - TestCase { - num: 8, - server_addr: "127.0.0.1:10000", - args: vec![ - "http://[::1]/d1", - "http://[::1]/d2", - "http://127.0.0.1/d3", - "http://127.0.0.1/d4", - ], - expected_err: Some(Error::other("all local endpoints should not have different hostnames/ips")), - ..Default::default() - }, - TestCase { - num: 9, - server_addr: "0.0.0.0:9001", - args: vec![ - "http://10.0.0.1:9000/export", - "http://10.0.0.2:9000/export", - case7_endpoint1.as_str(), - "http://10.0.0.2:9001/export", - ], - expected_err: Some(Error::other("same path '/export' can not be served by different port on same address")), - ..Default::default() - }, - TestCase { - num: 10, - server_addr: "0.0.0.0:9000", - args: vec![ - "http://127.0.0.1:9000/export", - case7_endpoint2.as_str(), - "http://10.0.0.1:9000/export", - "http://10.0.0.2:9000/export", - ], - expected_err: Some(Error::other("path '/export' cannot be served by different address on same server")), - ..Default::default() - }, - // DistErasure type - TestCase { - num: 11, - server_addr: "127.0.0.1:10000", - args: vec![ - case1_endpoint1.as_str(), - case1_endpoint2.as_str(), - "http://192.0.2.10/d3", - "http://192.0.2.11/d4", - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: case1_ur_ls[0].clone(), - is_local: case1_local_flags[0], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case1_ur_ls[1].clone(), - is_local: case1_local_flags[1], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case1_ur_ls[2].clone(), - is_local: case1_local_flags[2], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case1_ur_ls[3].clone(), - is_local: case1_local_flags[3], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::DistErasure), - ..Default::default() - }, - TestCase { - num: 12, - server_addr: "127.0.0.1:10000", - args: vec![ - case2_endpoint1.as_str(), - case2_endpoint2.as_str(), - "http://192.0.2.10/d3", - "http://192.0.2.11/d4", - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: case2_ur_ls[0].clone(), - is_local: case2_local_flags[0], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case2_ur_ls[1].clone(), - is_local: case2_local_flags[1], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case2_ur_ls[2].clone(), - is_local: case2_local_flags[2], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case2_ur_ls[3].clone(), - is_local: case2_local_flags[3], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::DistErasure), - ..Default::default() - }, - TestCase { - num: 13, - server_addr: "0.0.0.0:80", - args: vec![ - case3_endpoint1.as_str(), - "http://192.0.2.10:9000/d2", - "http://192.0.2.11/d3", - "http://192.0.2.12/d4", - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: case3_ur_ls[0].clone(), - is_local: case3_local_flags[0], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case3_ur_ls[1].clone(), - is_local: case3_local_flags[1], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case3_ur_ls[2].clone(), - is_local: case3_local_flags[2], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case3_ur_ls[3].clone(), - is_local: case3_local_flags[3], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::DistErasure), - ..Default::default() - }, - TestCase { - num: 14, - server_addr: "0.0.0.0:9000", - args: vec![ - case4_endpoint1.as_str(), - "http://192.0.2.10/d2", - "http://192.0.2.11/d3", - "http://192.0.2.12/d4", - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: case4_ur_ls[0].clone(), - is_local: case4_local_flags[0], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case4_ur_ls[1].clone(), - is_local: case4_local_flags[1], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case4_ur_ls[2].clone(), - is_local: case4_local_flags[2], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case4_ur_ls[3].clone(), - is_local: case4_local_flags[3], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::DistErasure), - ..Default::default() - }, - TestCase { - num: 15, - server_addr: "0.0.0.0:9000", - args: vec![ - case5_endpoint1.as_str(), - case5_endpoint2.as_str(), - case5_endpoint3.as_str(), - case5_endpoint4.as_str(), - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: case5_ur_ls[0].clone(), - is_local: case5_local_flags[0], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case5_ur_ls[1].clone(), - is_local: case5_local_flags[1], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case5_ur_ls[2].clone(), - is_local: case5_local_flags[2], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case5_ur_ls[3].clone(), - is_local: case5_local_flags[3], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::DistErasure), - ..Default::default() - }, - TestCase { - num: 16, - server_addr: "0.0.0.0:9003", - args: vec![ - "http://127.0.0.1:9000/d1", - "http://127.0.0.1:9001/d2", - "http://127.0.0.1:9002/d3", - case6_endpoint1.as_str(), - ], - expected_endpoints: Some(Endpoints(vec![ - Endpoint { - url: case6_ur_ls[0].clone(), - is_local: case6_local_flags[0], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case6_ur_ls[1].clone(), - is_local: case6_local_flags[1], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case6_ur_ls[2].clone(), - is_local: case6_local_flags[2], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - Endpoint { - url: case6_ur_ls[3].clone(), - is_local: case6_local_flags[3], - pool_idx: 0, - set_idx: 0, - disk_idx: 0, - }, - ])), - expected_setup_type: Some(SetupType::DistErasure), - ..Default::default() - }, - ]; - - for test_case in test_cases { - let disks_layout = match DisksLayout::from_volumes(test_case.args.as_slice()) { - Ok(v) => v, - Err(e) => { - if test_case.expected_err.is_none() { - panic!("Test {}: unexpected error: {}", test_case.num, e); - } - continue; - } - }; - - match ( - test_case.expected_err, - PoolEndpointList::create_pool_endpoints(test_case.server_addr, &disks_layout).await, - ) { - (None, Err(err)) => panic!("Test {}: error: expected = , got = {}", test_case.num, err), - (Some(err), Ok(_)) => panic!("Test {}: error: expected = {}, got = ", test_case.num, err), - (Some(e), Err(e2)) => { - assert_eq!( - e.to_string(), - e2.to_string(), - "Test {}: error: expected = {}, got = {}", - test_case.num, - e, - e2 - ) - } - (None, Ok(pools)) => { - if Some(&pools.setup_type) != test_case.expected_setup_type.as_ref() { - panic!( - "Test {}: setupType: expected = {:?}, got = {:?}", - test_case.num, test_case.expected_setup_type, &pools.setup_type - ) - } - - let left_len = test_case.expected_endpoints.as_ref().map(|v| v.as_ref().len()); - let right_len = pools.as_ref().first().map(|v| v.as_ref().len()); - - if left_len != right_len { - panic!("Test {}: endpoints len: expected = {:?}, got = {:?}", test_case.num, left_len, right_len); - } - - for (i, ep) in pools.as_ref()[0].as_ref().iter().enumerate() { - assert_eq!( - ep.to_string(), - test_case.expected_endpoints.as_ref().unwrap().as_ref()[i].to_string(), - "Test {}: endpoints: expected = {}, got = {}", - test_case.num, - test_case.expected_endpoints.as_ref().unwrap().as_ref()[i], - ep - ) - } - } - } - } - } - - fn must_file_path(s: impl AsRef) -> url::Url { - let path = s.as_ref().absolutize().expect("absolute test path"); - let url = url::Url::from_file_path(&path); - - assert!(url.is_ok(), "failed to convert path to URL: {}", path.display()); - - url.unwrap() - } - - fn must_url(s: &str) -> url::Url { - url::Url::parse(s).unwrap() - } - - fn get_expected_endpoints(args: Vec, prefix: String) -> (Vec, Vec) { - let mut urls = vec![]; - let mut local_flags = vec![]; - for arg in args { - urls.push(url::Url::parse(&arg).unwrap()); - local_flags.push(arg.starts_with(&prefix)); - } - - (urls, local_flags) - } - - #[tokio::test] - async fn test_create_server_endpoints() { - let test_cases = [ - // Invalid input. - ("", vec![], false), - // Range cannot be negative. - ("0.0.0.0:9000", vec!["/export1{-1...1}"], false), - // Range cannot start bigger than end. - ("0.0.0.0:9000", vec!["/export1{64...1}"], false), - // Range can only be numeric. - ("0.0.0.0:9000", vec!["/export1{a...z}"], false), - // Duplicate disks not allowed. - ("0.0.0.0:9000", vec!["/export1{1...32}", "/export1{1...32}"], false), - // Same host cannot export same disk on two ports - special case localhost. - ("0.0.0.0:9001", vec!["http://localhost:900{1...2}/export{1...64}"], false), - // Valid inputs. - ("0.0.0.0:9000", vec!["/export1"], true), - ("0.0.0.0:9000", vec!["/export1", "/export2", "/export3", "/export4"], true), - ("0.0.0.0:9000", vec!["/export1{1...64}"], true), - ("0.0.0.0:9000", vec!["/export1{01...64}"], true), - ("0.0.0.0:9000", vec!["/export1{1...32}", "/export1{33...64}"], true), - ("0.0.0.0:9001", vec!["http://localhost:9001/export{1...64}"], true), - ("0.0.0.0:9001", vec!["http://localhost:9001/export{01...64}"], true), - ]; - - for (i, test_case) in test_cases.iter().enumerate() { - let disks_layout = match DisksLayout::from_volumes(test_case.1.as_slice()) { - Ok(v) => v, - Err(e) => { - if test_case.2 { - panic!("Test {}: unexpected error: {}", i + 1, e); - } - continue; - } - }; - - let ret = EndpointServerPools::create_server_endpoints(test_case.0, &disks_layout).await; - - if let Err(err) = ret { - if test_case.2 { - panic!("Test {}: Expected success but failed instead {}", i + 1, err) - } - } else if !test_case.2 { - panic!("Test {}: expected failure but passed instead", i + 1); - } - } - } - - #[cfg(target_os = "linux")] - #[serial] - #[tokio::test] - async fn reject_shared_local_physical_disks_by_default() { - async_with_vars([(ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>), (ENV_MINIO_CI, None::<&str>)], async { - let dir = tempdir().unwrap(); - let disk1 = dir.path().join("disk1"); - let disk2 = dir.path().join("disk2"); - std::fs::create_dir_all(&disk1).unwrap(); - std::fs::create_dir_all(&disk2).unwrap(); - - let args = vec![disk1.to_string_lossy().into_owned(), disk2.to_string_lossy().into_owned()]; - let layout = DisksLayout::from_volumes(args.as_slice()).unwrap(); - - let err = EndpointServerPools::create_server_endpoints("0.0.0.0:9000", &layout) - .await - .unwrap_err(); - - let err_text = err.to_string(); - assert!(err_text.contains("distinct physical disks"), "unexpected error: {err_text}"); - assert!(err_text.contains(ENV_UNSAFE_BYPASS_DISK_CHECK), "unexpected error: {err_text}"); - assert!(err_text.contains("validation diagnostics:"), "unexpected error: {err_text}"); - assert!(err_text.contains("st_dev='"), "unexpected error: {err_text}"); - assert!(err_text.contains("device_ids=["), "unexpected error: {err_text}"); - }) - .await; - } - - #[cfg(target_os = "linux")] - #[serial] - #[tokio::test] - async fn allow_shared_local_physical_disks_with_explicit_env_bypass() { - async_with_vars([(ENV_UNSAFE_BYPASS_DISK_CHECK, Some("true"))], async { - let dir = tempdir().unwrap(); - let disk1 = dir.path().join("disk1"); - let disk2 = dir.path().join("disk2"); - std::fs::create_dir_all(&disk1).unwrap(); - std::fs::create_dir_all(&disk2).unwrap(); - - let args = vec![disk1.to_string_lossy().into_owned(), disk2.to_string_lossy().into_owned()]; - let layout = DisksLayout::from_volumes(args.as_slice()).unwrap(); - - let ret = EndpointServerPools::create_server_endpoints("0.0.0.0:9000", &layout).await; - assert!(ret.is_ok(), "expected bypassed disk validation to succeed, got {ret:?}"); - }) - .await; - } - - #[cfg(target_os = "linux")] - #[serial] - #[tokio::test] - async fn allow_shared_local_physical_disks_with_minio_ci_alias() { - async_with_vars([(ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>), (ENV_MINIO_CI, Some("1"))], async { - let dir = tempdir().unwrap(); - let disk1 = dir.path().join("disk1"); - let disk2 = dir.path().join("disk2"); - std::fs::create_dir_all(&disk1).unwrap(); - std::fs::create_dir_all(&disk2).unwrap(); - - let args = vec![disk1.to_string_lossy().into_owned(), disk2.to_string_lossy().into_owned()]; - let layout = DisksLayout::from_volumes(args.as_slice()).unwrap(); - - let ret = EndpointServerPools::create_server_endpoints("0.0.0.0:9000", &layout).await; - assert!(ret.is_ok(), "expected MINIO_CI alias to bypass disk validation, got {ret:?}"); - }) - .await; - } -} +pub use crate::layout::endpoints::*; diff --git a/crates/ecstore/src/layout/endpoint.rs b/crates/ecstore/src/layout/endpoint.rs new file mode 100644 index 000000000..6a8e86e3e --- /dev/null +++ b/crates/ecstore/src/layout/endpoint.rs @@ -0,0 +1,634 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::disk::error::{Error, Result}; +use path_absolutize::Absolutize; +use rustfs_utils::{is_local_host, is_socket_addr}; +use std::{fmt::Display, path::Path}; +use tracing::debug; +use url::{ParseError, Url}; + +#[cfg(windows)] +pub(crate) fn windows_fallback_local_path( + path: &str, + canonicalize_error: &std::io::Error, + context: &'static str, +) -> std::io::Result { + let absolute = Path::new(path).absolutize()?.to_path_buf(); + tracing::warn!( + path = %path, + canonicalize_error = %canonicalize_error, + resolved = ?absolute, + context = context, + "using windows fallback path resolution for local endpoint" + ); + Ok(absolute) +} + +/// enum for endpoint type. +#[derive(PartialEq, Eq, Debug)] +pub enum EndpointType { + /// path style endpoint type enum. + Path, + + /// URL style endpoint type enum. + Url, +} + +/// any type of endpoint. +#[derive(Debug, PartialEq, Eq, Clone, Hash)] +pub struct Endpoint { + pub url: Url, + pub is_local: bool, + + pub pool_idx: i32, + pub set_idx: i32, + pub disk_idx: i32, +} + +impl Display for Endpoint { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + if self.url.scheme() == "file" { + write!(f, "{}", self.get_file_path()) + } else { + write!(f, "{}", self.url) + } + } +} + +impl TryFrom<&str> for Endpoint { + /// The type returned in the event of a conversion error. + type Error = Error; + + /// Performs the conversion. + fn try_from(value: &str) -> std::result::Result { + // check whether given path is not empty. + if ["", "/", "\\"].iter().any(|&v| v.eq(value)) { + return Err(Error::other("empty or root endpoint is not supported")); + } + + let mut is_local = false; + let url = match Url::parse(value) { + #[allow(unused_mut)] + Ok(mut url) if url.has_host() => { + // URL style of endpoint. + // Valid URL style endpoint is + // - Scheme field must contain "http" or "https" + // - All field should be empty except Host and Path. + if !((url.scheme() == "http" || url.scheme() == "https") + && url.username().is_empty() + && url.fragment().is_none() + && url.query().is_none()) + { + return Err(Error::other("invalid URL endpoint format")); + } + + let path = url.path().to_string(); + + #[cfg(not(windows))] + let path = Path::new(&path).absolutize()?; + + #[cfg(windows)] + let path = if has_leading_slash_windows_drive(&path) { + // Url::path() exposes file-like Windows paths as `/C:/...`. + // Strip only that synthetic leading slash; plain URL paths + // such as `/export1` must stay URL paths, not become + // relative paths under the current drive. + Path::new(&path[1..]).absolutize()?.to_string_lossy().into_owned() + } else { + path + }; + #[cfg(windows)] + let path = Path::new(&path); + + debug!("endpoint try_from: path={}", path.display()); + + if path.parent().is_none() || path.as_os_str().is_empty() { + return Err(Error::other("empty or root path is not supported in URL endpoint")); + } + + match path.to_str() { + Some(v) => url.set_path(v), + None => return Err(Error::other("invalid path")), + } + + url + } + Ok(_) => { + // like d:/foo + is_local = true; + url_parse_from_file_path(value)? + } + Err(e) => match e { + ParseError::InvalidPort => { + return Err(Error::other("invalid URL endpoint format: port number must be between 1 to 65535")); + } + ParseError::EmptyHost => return Err(Error::other("invalid URL endpoint format: empty host name")), + ParseError::RelativeUrlWithoutBase => { + // like /foo + is_local = true; + url_parse_from_file_path(value)? + } + _ => return Err(Error::other(format!("invalid URL endpoint format: {e}"))), + }, + }; + + Ok(Endpoint { + url, + is_local, + pool_idx: -1, + set_idx: -1, + disk_idx: -1, + }) + } +} + +impl Endpoint { + /// returns type of endpoint. + pub fn get_type(&self) -> EndpointType { + if self.url.scheme() == "file" { + EndpointType::Path + } else { + EndpointType::Url + } + } + + /// sets a specific pool number to this node + pub fn set_pool_index(&mut self, idx: usize) { + self.pool_idx = idx as i32 + } + + /// sets a specific set number to this node + pub fn set_set_index(&mut self, idx: usize) { + self.set_idx = idx as i32 + } + + /// sets a specific disk number to this node + pub fn set_disk_index(&mut self, idx: usize) { + self.disk_idx = idx as i32 + } + + /// resolves the host and updates if it is local or not. + pub fn update_is_local(&mut self, local_port: u16) -> Result<()> { + match (self.url.scheme(), self.url.host()) { + (v, Some(host)) if v != "file" => { + self.is_local = is_local_host(host, self.url.port().unwrap_or_default(), local_port)?; + } + _ => {} + } + + Ok(()) + } + + /// returns the host to be used for grid connections. + pub fn grid_host(&self) -> String { + match (self.url.host(), self.url.port()) { + (Some(host), Some(port)) => { + debug!("grid_host scheme={}: host={}, port={}", self.url.scheme(), host, port); + format!("{}://{}:{}", self.url.scheme(), host, port) + } + (Some(host), None) => { + debug!("grid_host scheme={}: host={}", self.url.scheme(), host); + format!("{}://{}", self.url.scheme(), host) + } + _ => String::new(), + } + } + + pub fn host_port(&self) -> String { + match (self.url.host(), self.url.port()) { + (Some(host), Some(port)) => { + debug!("host_port host={}, port={}", host, port); + format!("{host}:{port}") + } + (Some(host), None) => { + debug!("host_port host={}, port={}", host, self.url.port().unwrap_or(0)); + format!("{host}") + } + _ => String::new(), + } + } + + pub fn get_file_path(&self) -> String { + let path: &str = self.url.path(); + let decoded: std::borrow::Cow<'_, str> = match urlencoding::decode(path) { + Ok(decoded) => decoded, + Err(e) => { + debug!("Failed to decode path '{}': {}, using original path", path, e); + std::borrow::Cow::Borrowed(path) + } + }; + #[cfg(windows)] + if self.url.scheme() == "file" { + let stripped: &str = decoded.strip_prefix('/').unwrap_or(&decoded); + debug!("get_file_path windows: path={}", stripped); + return stripped.to_string(); + } + decoded.into_owned() + } +} + +#[cfg(windows)] +fn has_leading_slash_windows_drive(path: &str) -> bool { + let bytes = path.as_bytes(); + bytes.len() >= 4 && bytes[0] == b'/' && bytes[1].is_ascii_alphabetic() && bytes[2] == b':' && bytes[3] == b'/' +} + +/// parse a file path into a URL. +fn url_parse_from_file_path(value: &str) -> Result { + // Only check if the arg is an ip address and ask for scheme since its absent. + // localhost, example.com, any FQDN cannot be disambiguated from a regular file path such as + // /mnt/export1. So we go ahead and start the rustfs server in FS modes in these cases. + let addr: Vec<&str> = value.splitn(2, '/').collect(); + if is_socket_addr(addr[0]) { + return Err(Error::other("invalid URL endpoint format: missing scheme http or https")); + } + + let file_path = match Path::new(value).absolutize() { + Ok(path) => path, + Err(err) => return Err(Error::other(format!("absolute path failed: {err}"))), + }; + + match Url::from_file_path(file_path) { + Ok(url) => Ok(url), + Err(_) => Err(Error::other("Convert a file path into an URL failed")), + } +} + +#[cfg(test)] +mod test { + use super::*; + + fn expected_file_path(path: &str) -> String { + Path::new(path).absolutize().unwrap().to_string_lossy().replace('\\', "/") + } + + fn expected_file_url(path: &str) -> Url { + url_parse_from_file_path(path).unwrap() + } + + #[test] + fn test_new_endpoint() { + #[derive(Default)] + struct TestCase<'a> { + arg: &'a str, + expected_endpoint: Option, + expected_type: Option, + expected_err: Option, + } + + let u2 = Url::parse("https://example.org/path").unwrap(); + let u4 = Url::parse("http://192.168.253.200/path").unwrap(); + let u6 = Url::parse("http://server:/path").unwrap(); + let root_slash_foo = expected_file_url("/foo"); + + let test_cases = [ + TestCase { + arg: "/foo", + expected_endpoint: Some(Endpoint { + url: root_slash_foo, + is_local: true, + pool_idx: -1, + set_idx: -1, + disk_idx: -1, + }), + expected_type: Some(EndpointType::Path), + expected_err: None, + }, + TestCase { + arg: "https://example.org/path", + expected_endpoint: Some(Endpoint { + url: u2, + is_local: false, + pool_idx: -1, + set_idx: -1, + disk_idx: -1, + }), + expected_type: Some(EndpointType::Url), + expected_err: None, + }, + TestCase { + arg: "http://192.168.253.200/path", + expected_endpoint: Some(Endpoint { + url: u4, + is_local: false, + pool_idx: -1, + set_idx: -1, + disk_idx: -1, + }), + expected_type: Some(EndpointType::Url), + expected_err: None, + }, + TestCase { + arg: "", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("empty or root endpoint is not supported")), + }, + TestCase { + arg: "/", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("empty or root endpoint is not supported")), + }, + TestCase { + arg: "\\", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("empty or root endpoint is not supported")), + }, + TestCase { + arg: "c://foo", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format")), + }, + TestCase { + arg: "ftp://foo", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format")), + }, + TestCase { + arg: "http://server/path?location", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format")), + }, + TestCase { + arg: "http://:/path", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format: empty host name")), + }, + TestCase { + arg: "http://:8080/path", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format: empty host name")), + }, + TestCase { + arg: "http://server:/path", + expected_endpoint: Some(Endpoint { + url: u6, + is_local: false, + pool_idx: -1, + set_idx: -1, + disk_idx: -1, + }), + expected_type: Some(EndpointType::Url), + expected_err: None, + }, + TestCase { + arg: "https://93.184.216.34:808080/path", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format: port number must be between 1 to 65535")), + }, + TestCase { + arg: "http://server:8080//", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("empty or root path is not supported in URL endpoint")), + }, + TestCase { + arg: "http://server:8080/", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("empty or root path is not supported in URL endpoint")), + }, + TestCase { + arg: "192.168.1.210:9000", + expected_endpoint: None, + expected_type: None, + expected_err: Some(Error::other("invalid URL endpoint format: missing scheme http or https")), + }, + ]; + + for test_case in test_cases { + let ret = Endpoint::try_from(test_case.arg); + if test_case.expected_err.is_none() && ret.is_err() { + panic!("{}: error: expected = , got = {:?}", test_case.arg, ret); + } + if test_case.expected_err.is_some() && ret.is_ok() { + panic!("{}: error: expected = {:?}, got = ", test_case.arg, test_case.expected_err); + } + match (test_case.expected_err, ret) { + (None, Err(e)) => panic!("{}: error: expected = , got = {}", test_case.arg, e), + (None, Ok(mut ep)) => { + let _ = ep.update_is_local(9000); + if test_case.expected_type != Some(ep.get_type()) { + panic!( + "{}: type: expected = {:?}, got = {:?}", + test_case.arg, + test_case.expected_type, + ep.get_type() + ); + } + + assert_eq!(test_case.expected_endpoint, Some(ep), "{}: endpoint", test_case.arg); + } + (Some(e), Ok(_)) => panic!("{}: error: expected = {}, got = ", test_case.arg, e), + (Some(e), Err(e2)) => { + assert_eq!(e.to_string(), e2.to_string(), "{}: error: expected = {}, got = {}", test_case.arg, e, e2) + } + } + } + } + + #[test] + fn test_endpoint_display() { + // Test file path display + let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); + let display_str = format!("{file_endpoint}"); + assert_eq!(display_str, expected_file_path("/tmp/data")); + + // Test URL display + let url_endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); + let display_str = format!("{url_endpoint}"); + assert_eq!(display_str, "http://example.com:9000/path"); + } + + #[test] + fn test_endpoint_type() { + let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); + assert_eq!(file_endpoint.get_type(), EndpointType::Path); + + let url_endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); + assert_eq!(url_endpoint.get_type(), EndpointType::Url); + } + + #[test] + fn test_endpoint_indexes() { + let mut endpoint = Endpoint::try_from("/tmp/data").unwrap(); + + // Test initial values + assert_eq!(endpoint.pool_idx, -1); + assert_eq!(endpoint.set_idx, -1); + assert_eq!(endpoint.disk_idx, -1); + + // Test setting indexes + endpoint.set_pool_index(2); + endpoint.set_set_index(3); + endpoint.set_disk_index(4); + + assert_eq!(endpoint.pool_idx, 2); + assert_eq!(endpoint.set_idx, 3); + assert_eq!(endpoint.disk_idx, 4); + } + + #[test] + fn test_endpoint_grid_host() { + let endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); + assert_eq!(endpoint.grid_host(), "http://example.com:9000"); + + let endpoint_no_port = Endpoint::try_from("https://example.com/path").unwrap(); + assert_eq!(endpoint_no_port.grid_host(), "https://example.com"); + + let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); + assert_eq!(file_endpoint.grid_host(), ""); + } + + #[test] + fn test_endpoint_host_port() { + let endpoint = Endpoint::try_from("http://example.com:9000/path").unwrap(); + assert_eq!(endpoint.host_port(), "example.com:9000"); + + let endpoint_no_port = Endpoint::try_from("https://example.com/path").unwrap(); + assert_eq!(endpoint_no_port.host_port(), "example.com"); + + let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); + assert_eq!(file_endpoint.host_port(), ""); + } + + #[test] + fn test_endpoint_get_file_path() { + let file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); + assert_eq!(file_endpoint.get_file_path(), expected_file_path("/tmp/data")); + + let url_endpoint = Endpoint::try_from("http://example.com:9000/path/to/data").unwrap(); + assert_eq!(url_endpoint.get_file_path(), "/path/to/data"); + } + + #[cfg(windows)] + #[test] + fn test_windows_url_drive_path_requires_separator_after_colon() { + let drive_path_endpoint = Endpoint::try_from("http://host/C:/data").unwrap(); + assert_eq!(drive_path_endpoint.get_type(), EndpointType::Url); + assert!(has_leading_slash_windows_drive(Url::parse("http://host/C:/data").unwrap().path())); + + let url_path_endpoint = Endpoint::try_from("http://host/C:foo").unwrap(); + assert_eq!(url_path_endpoint.get_type(), EndpointType::Url); + assert!(!has_leading_slash_windows_drive(Url::parse("http://host/C:foo").unwrap().path())); + assert_eq!(url_path_endpoint.get_file_path(), "/C:foo"); + } + + #[test] + fn test_endpoint_clone_and_equality() { + let endpoint1 = Endpoint::try_from("/tmp/data").unwrap(); + let endpoint2 = endpoint1.clone(); + + assert_eq!(endpoint1, endpoint2); + assert_eq!(endpoint1.url, endpoint2.url); + assert_eq!(endpoint1.is_local, endpoint2.is_local); + assert_eq!(endpoint1.pool_idx, endpoint2.pool_idx); + assert_eq!(endpoint1.set_idx, endpoint2.set_idx); + assert_eq!(endpoint1.disk_idx, endpoint2.disk_idx); + } + + #[test] + fn test_endpoint_with_special_paths() { + // Test with complex paths + let complex_path = "/var/lib/rustfs/data/bucket1"; + let endpoint = Endpoint::try_from(complex_path).unwrap(); + assert_eq!(endpoint.get_file_path(), expected_file_path(complex_path)); + assert!(endpoint.is_local); + assert_eq!(endpoint.get_type(), EndpointType::Path); + } + + #[test] + fn test_endpoint_with_spaces_in_path() { + let path_with_spaces = "/Users/test/Library/Application Support/rustfs/data"; + let endpoint = Endpoint::try_from(path_with_spaces).unwrap(); + assert_eq!(endpoint.get_file_path(), expected_file_path(path_with_spaces)); + assert!(endpoint.is_local); + assert_eq!(endpoint.get_type(), EndpointType::Path); + } + + #[test] + fn test_endpoint_percent_encoding_roundtrip() { + let path_with_spaces = "/Users/test/Library/Application Support/rustfs/data"; + let endpoint = Endpoint::try_from(path_with_spaces).unwrap(); + + // Verify that the URL internally stores percent-encoded path + assert!( + endpoint.url.path().contains("%20"), + "URL path should contain percent-encoded spaces: {}", + endpoint.url.path() + ); + + // Verify that get_file_path() decodes the percent-encoded path correctly + assert_eq!( + endpoint.get_file_path(), + expected_file_path("/Users/test/Library/Application Support/rustfs/data"), + "get_file_path() should decode percent-encoded spaces" + ); + } + + #[test] + fn test_endpoint_with_various_special_characters() { + // Test path with multiple special characters that get percent-encoded + let path_with_special = "/tmp/test path/data[1]/file+name&more"; + let endpoint = Endpoint::try_from(path_with_special).unwrap(); + + // get_file_path() should return the original path with decoded characters + assert_eq!(endpoint.get_file_path(), expected_file_path(path_with_special)); + } + + #[test] + fn test_endpoint_update_is_local() { + let mut endpoint = Endpoint::try_from("http://localhost:9000/path").unwrap(); + let result = endpoint.update_is_local(9000); + assert!(result.is_ok()); + + let mut file_endpoint = Endpoint::try_from("/tmp/data").unwrap(); + let result = file_endpoint.update_is_local(9000); + assert!(result.is_ok()); + } + + #[test] + fn test_url_parse_from_file_path() { + let result = url_parse_from_file_path("/tmp/test"); + assert!(result.is_ok()); + + let url = result.unwrap(); + assert_eq!(url.scheme(), "file"); + } + + #[test] + fn test_endpoint_hash() { + use std::collections::HashSet; + + let endpoint1 = Endpoint::try_from("/tmp/data1").unwrap(); + let endpoint2 = Endpoint::try_from("/tmp/data2").unwrap(); + let endpoint3 = endpoint1.clone(); + + let mut set = HashSet::new(); + set.insert(endpoint1); + set.insert(endpoint2); + set.insert(endpoint3); // Should not be added as it's equal to endpoint1 + + assert_eq!(set.len(), 2); + } +} diff --git a/crates/ecstore/src/layout/endpoints.rs b/crates/ecstore/src/layout/endpoints.rs new file mode 100644 index 000000000..3c979f0fd --- /dev/null +++ b/crates/ecstore/src/layout/endpoints.rs @@ -0,0 +1,1698 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use crate::{ + global::global_rustfs_port, + layout::{ + disks_layout::DisksLayout, + endpoint::{Endpoint, EndpointType}, + }, +}; +use rustfs_config::{DEFAULT_UNSAFE_BYPASS_DISK_CHECK, ENV_MINIO_CI, ENV_UNSAFE_BYPASS_DISK_CHECK}; +use rustfs_utils::{XHost, check_local_server_addr, get_host_ip, is_local_host}; +use std::{ + collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry}, + io::{Error, ErrorKind, Result}, + net::IpAddr, +}; +use tracing::{error, info, instrument, warn}; + +/// enum for setup type. +#[derive(PartialEq, Eq, Debug, Clone)] +pub enum SetupType { + /// starts with unknown setup type. + Unknown, + + /// FS setup type enum. + FS, + + /// Erasure single drive setup enum. + ErasureSD, + + /// Erasure setup type enum. + Erasure, + + /// Distributed Erasure setup type enum. + DistErasure, +} + +/// holds information about a node in this cluster +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Node { + pub url: url::Url, + pub pools: Vec, + pub is_local: bool, + pub grid_host: String, +} + +/// list of same type of endpoint. +#[derive(Debug, Default, Clone)] +pub struct Endpoints(Vec); + +#[derive(Debug, Clone)] +struct LocalDiskValidationDiagnostic { + original_path: String, + canonical_path: Option, + device_numbers: Option, + device_ids: Option>, +} + +impl LocalDiskValidationDiagnostic { + fn new(original_path: &str) -> Self { + Self { + original_path: original_path.to_string(), + canonical_path: None, + device_numbers: None, + device_ids: None, + } + } + + fn summary(&self) -> String { + let canonical_path = self.canonical_path.as_deref().unwrap_or("(unresolved)"); + let device_numbers = self.device_numbers.as_deref().unwrap_or("(unavailable)"); + let device_ids = self + .device_ids + .as_ref() + .map(|ids| ids.join(",")) + .unwrap_or_else(|| "(unavailable)".to_string()); + + format!( + "path='{}', canonical='{}', st_dev='{}', device_ids=[{}]", + self.original_path, canonical_path, device_numbers, device_ids + ) + } +} + +impl AsRef> for Endpoints { + fn as_ref(&self) -> &Vec { + &self.0 + } +} + +impl AsMut> for Endpoints { + fn as_mut(&mut self) -> &mut Vec { + &mut self.0 + } +} + +impl From> for Endpoints { + fn from(v: Vec) -> Self { + Self(v) + } +} + +impl> TryFrom<&[T]> for Endpoints { + type Error = Error; + + /// returns new endpoint list based on input args. + fn try_from(args: &[T]) -> Result { + let mut endpoint_type = None; + let mut schema = None; + let mut endpoints = Vec::with_capacity(args.len()); + let mut uniq_set = HashSet::with_capacity(args.len()); + + // Loop through args and adds to endpoint list. + for (i, arg) in args.iter().enumerate() { + let endpoint = match Endpoint::try_from(arg.as_ref()) { + Ok(ep) => ep, + Err(e) => return Err(Error::other(format!("'{}': {}", arg.as_ref(), e))), + }; + + // All endpoints have to be same type and scheme if applicable. + if i == 0 { + endpoint_type = Some(endpoint.get_type()); + schema = Some(endpoint.url.scheme().to_owned()); + } else if Some(endpoint.get_type()) != endpoint_type { + return Err(Error::other("mixed style endpoints are not supported")); + } else if Some(endpoint.url.scheme()) != schema.as_deref() { + return Err(Error::other("mixed scheme is not supported")); + } + + // Check for duplicate endpoints. + let endpoint_str = endpoint.to_string(); + if uniq_set.contains(&endpoint_str) { + return Err(Error::other("duplicate endpoints found")); + } + + uniq_set.insert(endpoint_str); + endpoints.push(endpoint); + } + + Ok(Endpoints(endpoints)) + } +} + +impl Endpoints { + /// Converts `self` into its inner representation. + /// + /// This method consumes the `self` object and returns its inner `Vec`. + /// It is useful for when you need to take the endpoints out of their container + /// without needing a reference to the container itself. + pub fn into_inner(self) -> Vec { + self.0 + } + + pub fn into_ref(&self) -> &Vec { + &self.0 + } + + // GetString - returns endpoint string of i-th endpoint (0-based), + // and empty string for invalid indexes. + pub fn get_string(&self, i: usize) -> String { + if i >= self.0.len() { + return "".to_string(); + } + + self.0[i].to_string() + } + + pub fn is_empty(&self) -> bool { + self.0.is_empty() + } +} + +#[derive(Debug)] +/// a temporary type to holds the list of endpoints +struct PoolEndpointList { + inner: Vec, + setup_type: SetupType, +} + +impl AsRef> for PoolEndpointList { + fn as_ref(&self) -> &Vec { + &self.inner + } +} + +impl AsMut> for PoolEndpointList { + fn as_mut(&mut self) -> &mut Vec { + &mut self.inner + } +} + +impl PoolEndpointList { + /// creates a list of endpoints per pool, resolves their relevant + /// hostnames and discovers those are local or remote. + async fn create_pool_endpoints(server_addr: &str, disks_layout: &DisksLayout) -> Result { + if disks_layout.is_empty_layout() { + return Err(Error::other("invalid number of endpoints")); + } + + let server_addr = check_local_server_addr(server_addr)?; + + // For single arg, return single drive EC setup. + if disks_layout.is_single_drive_layout() { + let mut endpoint = Endpoint::try_from(disks_layout.get_single_drive_layout())?; + endpoint.update_is_local(server_addr.port())?; + + if endpoint.get_type() != EndpointType::Path { + return Err(Error::other("use path style endpoint for single node setup")); + } + + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(0); + + // TODO Check for cross device mounts if any. + + return Ok(Self { + inner: vec![Endpoints::from(vec![endpoint])], + setup_type: SetupType::ErasureSD, + }); + } + + let mut pool_endpoints = Vec::::with_capacity(disks_layout.pools.len()); + for (pool_idx, pool) in disks_layout.pools.iter().enumerate() { + let mut endpoints = Endpoints::default(); + for (set_idx, set_layout) in pool.iter().enumerate() { + // Convert args to endpoints + let mut eps = Endpoints::try_from(set_layout.as_slice())?; + + // TODO Check for cross device mounts if any. + + for (disk_idx, ep) in eps.as_mut().iter_mut().enumerate() { + ep.set_pool_index(pool_idx); + ep.set_set_index(set_idx); + ep.set_disk_index(disk_idx); + } + + endpoints.as_mut().append(eps.as_mut()); + } + + if endpoints.as_ref().is_empty() { + return Err(Error::other("invalid number of endpoints")); + } + + pool_endpoints.push(endpoints); + } + + // setup type + let mut unique_args = HashSet::new(); + let mut pool_endpoint_list = Self { + inner: pool_endpoints, + setup_type: SetupType::Unknown, + }; + + pool_endpoint_list.update_is_local(server_addr.port())?; + + for endpoints in pool_endpoint_list.inner.iter_mut() { + // Check whether same path is not used in endpoints of a host on different port. + let mut path_ip_map: HashMap> = HashMap::new(); + let mut host_ip_cache = HashMap::new(); + for ep in endpoints.as_ref() { + if !ep.url.has_host() { + continue; + } + + let host = ep.url.host().unwrap(); + let host_ip_set = if let Some(set) = host_ip_cache.get(&host) { + info!( + target: "rustfs::ecstore::endpoints", + host = %host, + endpoint = %ep.to_string(), + from = "cache", + "Create pool endpoints host '{}' found in cache for endpoint '{}'", host, ep.to_string() + ); + set + } else { + let ips = match get_host_ip(host.clone()).await { + Ok(ips) => ips, + Err(e) => { + error!("Create pool endpoints host {} not found, error:{}", host, e); + return Err(Error::other(format!("host '{host}' cannot resolve: {e}"))); + } + }; + info!( + target: "rustfs::ecstore::endpoints", + host = %host, + endpoint = %ep.to_string(), + from = "get_host_ip", + "Create pool endpoints host '{}' resolved to ips {:?} for endpoint '{}'", + host, + ips, + ep.to_string() + ); + host_ip_cache.insert(host.clone(), ips); + host_ip_cache.get(&host).unwrap() + }; + + let path = ep.get_file_path(); + match path_ip_map.entry(path) { + Entry::Occupied(mut e) => { + if e.get().intersection(host_ip_set).count() > 0 { + let path_key = e.key().clone(); + return Err(Error::other(format!( + "same path '{path_key}' can not be served by different port on same address" + ))); + } + e.get_mut().extend(host_ip_set.iter()); + } + Entry::Vacant(e) => { + e.insert(host_ip_set.clone()); + } + } + } + + // Check whether same path is used for more than 1 local endpoints. + let mut local_path_set = HashSet::new(); + for ep in endpoints.as_ref() { + if !ep.is_local { + continue; + } + + let path = ep.get_file_path(); + if local_path_set.contains(&path) { + return Err(Error::other(format!( + "path '{path}' cannot be served by different address on same server" + ))); + } + local_path_set.insert(path); + } + + // Here all endpoints are URL style. + let mut ep_path_set = HashSet::new(); + let mut local_server_host_set = HashSet::new(); + let mut local_port_set = HashSet::new(); + let mut local_endpoint_count = 0; + + for ep in endpoints.as_ref() { + ep_path_set.insert(ep.get_file_path()); + if ep.is_local && ep.url.has_host() { + local_server_host_set.insert(ep.url.host()); + local_port_set.insert(ep.url.port()); + local_endpoint_count += 1; + } + } + + // All endpoints are pointing to local host + if endpoints.as_ref().len() == local_endpoint_count { + // If all endpoints have same port number, Just treat it as local erasure setup + // using URL style endpoints. + if local_port_set.len() == 1 && local_server_host_set.len() > 1 { + return Err(Error::other("all local endpoints should not have different hostnames/ips")); + } + } + + // Add missing port in all endpoints. + for ep in endpoints.as_mut() { + if !ep.url.has_host() { + unique_args.insert(format!("localhost:{}", server_addr.port())); + continue; + } + match ep.url.port() { + None => { + let _ = ep.url.set_port(Some(server_addr.port())); + } + Some(port) => { + // If endpoint is local, but port is different than serverAddrPort, then make it as remote. + if ep.is_local && server_addr.port() != port { + ep.is_local = false; + } + } + } + unique_args.insert(ep.host_port()); + } + } + + validate_local_physical_disk_independence(pool_endpoint_list.as_ref())?; + + let setup_type = match pool_endpoint_list.as_ref()[0].as_ref()[0].get_type() { + EndpointType::Path => SetupType::Erasure, + EndpointType::Url => match unique_args.len() { + 1 => SetupType::Erasure, + _ => SetupType::DistErasure, + }, + }; + + pool_endpoint_list.setup_type = setup_type; + + Ok(pool_endpoint_list) + } + + /// resolves all hosts and discovers which are local + fn update_is_local(&mut self, local_port: u16) -> Result<()> { + for endpoints in self.inner.iter_mut() { + for ep in endpoints.as_mut() { + match ep.url.host() { + None => { + ep.is_local = true; + } + Some(host) => { + ep.is_local = is_local_host(host, ep.url.port().unwrap_or_default(), local_port)?; + } + } + } + } + + Ok(()) + } + + /// resolves all hosts and discovers which are local + fn _update_is_local(&mut self, local_port: u16) -> Result<()> { + let mut eps_resolved = 0; + let mut found_local = false; + let mut resolved_set: HashSet<(usize, usize)> = HashSet::new(); + let ep_count: usize = self.inner.iter().map(|v| v.as_ref().len()).sum(); + + loop { + // Break if the local endpoint is found already Or all the endpoints are resolved. + if found_local || eps_resolved == ep_count { + break; + } + + for (i, endpoints) in self.inner.iter_mut().enumerate() { + for (j, ep) in endpoints.as_mut().iter_mut().enumerate() { + if resolved_set.contains(&(i, j)) { + // Continue if host is already resolved. + continue; + } + + match ep.url.host() { + None => { + if !found_local { + found_local = true; + } + ep.is_local = true; + eps_resolved += 1; + resolved_set.insert((i, j)); + continue; + } + Some(host) => match is_local_host(host, ep.url.port().unwrap_or_default(), local_port) { + Ok(is_local) => { + if !found_local { + found_local = is_local; + } + ep.is_local = is_local; + eps_resolved += 1; + resolved_set.insert((i, j)); + } + Err(err) => { + // TODO Retry infinitely on Kubernetes and Docker swarm? + return Err(err); + } + }, + } + } + } + } + + Ok(()) + } +} + +/// represent endpoints in a given pool +/// along with its setCount and setDriveCount. +#[derive(Debug, Clone)] +pub struct PoolEndpoints { + // indicates if endpoints are provided in non-ellipses style + pub legacy: bool, + pub set_count: usize, + pub drives_per_set: usize, + pub endpoints: Endpoints, + pub cmd_line: String, + pub platform: String, +} + +/// list of endpoints +#[derive(Debug, Clone, Default)] +pub struct EndpointServerPools(pub Vec); + +impl From> for EndpointServerPools { + fn from(v: Vec) -> Self { + Self(v) + } +} + +impl AsRef> for EndpointServerPools { + fn as_ref(&self) -> &Vec { + &self.0 + } +} + +impl AsMut> for EndpointServerPools { + fn as_mut(&mut self) -> &mut Vec { + &mut self.0 + } +} + +impl EndpointServerPools { + pub fn reset(&mut self, eps: Vec) { + self.0 = eps; + } + pub fn legacy(&self) -> bool { + self.0.len() == 1 && self.0[0].legacy + } + pub fn get_pool_idx(&self, cmd_line: &str) -> Option { + for (idx, eps) in self.0.iter().enumerate() { + if eps.cmd_line.as_str() == cmd_line { + return Some(idx); + } + } + None + } + pub async fn from_volumes(server_addr: &str, endpoints: Vec) -> Result<(EndpointServerPools, SetupType)> { + let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; + + Self::create_server_endpoints(server_addr, &layouts).await + } + /// validates and creates new endpoints from input args, supports + /// both ellipses and without ellipses transparently. + pub async fn create_server_endpoints( + server_addr: &str, + disks_layout: &DisksLayout, + ) -> Result<(EndpointServerPools, SetupType)> { + if disks_layout.pools.is_empty() { + return Err(Error::other("Invalid arguments specified")); + } + + let pool_eps = PoolEndpointList::create_pool_endpoints(server_addr, disks_layout).await?; + + let mut ret: EndpointServerPools = Vec::with_capacity(pool_eps.as_ref().len()).into(); + for (i, eps) in pool_eps.inner.into_iter().enumerate() { + let ep = PoolEndpoints { + legacy: disks_layout.legacy, + set_count: disks_layout.get_set_count(i), + drives_per_set: disks_layout.get_drives_per_set(i), + endpoints: eps, + cmd_line: disks_layout.get_cmd_line(i), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }; + + ret.add(ep)?; + } + + Ok((ret, pool_eps.setup_type)) + } + + pub fn es_count(&self) -> usize { + self.0.iter().map(|v| v.set_count).sum() + } + + /// add pool endpoints + pub fn add(&mut self, eps: PoolEndpoints) -> Result<()> { + let mut exits = HashSet::new(); + for peps in self.0.iter() { + for ep in peps.endpoints.as_ref() { + exits.insert(ep.to_string()); + } + } + + for ep in eps.endpoints.as_ref() { + if exits.contains(&ep.to_string()) { + return Err(Error::other("duplicate endpoints found")); + } + } + + self.0.push(eps); + + Ok(()) + } + + /// returns true if the first endpoint is local. + pub fn first_local(&self) -> bool { + self.0 + .first() + .and_then(|v| v.endpoints.as_ref().first()) + .is_some_and(|v| v.is_local) + } + + /// returns a sorted list of nodes in this cluster + pub fn get_nodes(&self) -> Vec { + let mut node_map = HashMap::new(); + + for pool in self.0.iter() { + for ep in pool.endpoints.as_ref() { + let n = node_map.entry(ep.host_port()).or_insert_with(|| Node { + url: ep.url.clone(), + pools: vec![], + is_local: ep.is_local, + grid_host: ep.grid_host(), + }); + + if !n.pools.contains(&(ep.pool_idx as usize)) { + n.pools.push(ep.pool_idx as usize); + } + } + } + + let mut nodes: Vec = node_map.into_values().collect(); + + nodes.sort_by(|a, b| a.grid_host.cmp(&b.grid_host)); + + nodes + } + + #[instrument] + pub fn hosts_sorted(&self) -> Vec> { + let (mut peers, local) = self.peers(); + + let mut ret = vec![None; peers.len()]; + + peers.sort(); + + for (i, peer) in peers.iter().enumerate() { + if &local == peer { + continue; + } + + let host = match XHost::try_from(peer.clone()) { + Ok(res) => res, + Err(err) => { + warn!("Xhost parse failed {:?}", err); + continue; + } + }; + + ret[i] = Some(host); + } + + ret + } + pub fn peers(&self) -> (Vec, 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 && endpoint.url.port() == Some(global_rustfs_port()) && local.is_none() { + local = Some(host.clone()); + } + + set.insert(host); + } + } + + let hosts: Vec = set.iter().cloned().collect(); + + (hosts, local.unwrap_or_default()) + } + + pub fn find_grid_hosts_from_peer(&self, host: &XHost) -> Option { + for ep in self.0.iter() { + for endpoint in ep.endpoints.0.iter() { + if endpoint.is_local { + continue; + } + let xhost = match XHost::try_from(endpoint.host_port()) { + Ok(res) => res, + Err(_) => { + continue; + } + }; + + if xhost.to_string() == host.to_string() { + return Some(endpoint.grid_host()); + } + } + } + + None + } +} + +fn validate_local_physical_disk_independence(pools: &[Endpoints]) -> Result<()> { + let mut local_paths = BTreeSet::new(); + for endpoints in pools { + for endpoint in endpoints.as_ref() { + if endpoint.is_local { + local_paths.insert(endpoint.get_file_path()); + } + } + } + + if local_paths.is_empty() { + return Ok(()); + } + + let local_paths = local_paths.into_iter().collect::>(); + validate_local_cross_device_mounts(&local_paths)?; + + if local_paths.len() <= 1 { + return Ok(()); + } + + // Compatibility behavior: + // - canonical key: RUSTFS_UNSAFE_BYPASS_DISK_CHECK + // - legacy CI alias: MINIO_CI + // If both are set, `get_env_bool_with_aliases` keeps canonical key precedence. + if rustfs_utils::get_env_bool_with_aliases(ENV_UNSAFE_BYPASS_DISK_CHECK, &[ENV_MINIO_CI], DEFAULT_UNSAFE_BYPASS_DISK_CHECK) { + warn!( + env = ENV_UNSAFE_BYPASS_DISK_CHECK, + local_paths = ?local_paths, + "Skipping local physical disk independence validation due to explicit environment override", + ); + return Ok(()); + } + + let mut device_paths = BTreeMap::>::new(); + let mut diagnostics = Vec::with_capacity(local_paths.len()); + #[cfg(not(windows))] + let mut missing_paths = Vec::new(); + + for path in &local_paths { + let mut diagnostic = LocalDiskValidationDiagnostic::new(path); + let canonical = match rustfs_utils::canonicalize(path) { + Ok(path) => path, + Err(err) if err.kind() == ErrorKind::NotFound => { + // On Windows, canonicalize can fail for ZFS volumes, junction points, + // subst drives, and other non-standard mounts. Try absolutize as fallback. + #[cfg(windows)] + { + match crate::layout::endpoint::windows_fallback_local_path(path, &err, "disk independence validation") { + Ok(absolute) => { + let abs_path = absolute.to_string_lossy().into_owned(); + diagnostic.canonical_path = Some(abs_path.clone()); + if let Ok(serial) = rustfs_utils::os::get_volume_serial_number(&abs_path) { + diagnostic.device_numbers = Some(format!("serial:{serial:#010x}")); + } + match rustfs_utils::os::get_physical_device_ids(&abs_path) { + Ok(ids) => { + diagnostic.device_ids = Some(ids.clone()); + for device_id in ids { + device_paths.entry(device_id).or_default().insert(abs_path.clone()); + } + } + Err(device_err) => { + return Err(Error::other(format!( + "failed to inspect physical disk for local endpoint '{abs_path}' after fallback path resolution: {device_err}" + ))); + } + } + diagnostics.push(diagnostic); + continue; + } + Err(fallback_err) => { + return Err(Error::other(format!( + "failed to resolve local endpoint path '{path}' for disk validation: {err}; fallback resolution failed: {fallback_err}" + ))); + } + } + } + #[cfg(not(windows))] + { + missing_paths.push(path.clone()); + diagnostics.push(diagnostic); + continue; + } + } + Err(err) => { + return Err(Error::other(format!( + "failed to resolve local endpoint path '{path}' for disk validation: {err}" + ))); + } + }; + let canonical_path = canonical.to_string_lossy().into_owned(); + diagnostic.canonical_path = Some(canonical_path.clone()); + #[cfg(not(windows))] + if let Ok(stat) = rustix::fs::stat(canonical.as_path()) { + diagnostic.device_numbers = Some(format!("{}:{}", rustix::fs::major(stat.st_dev), rustix::fs::minor(stat.st_dev))); + } + #[cfg(windows)] + if let Ok(serial) = rustfs_utils::os::get_volume_serial_number(&canonical_path) { + diagnostic.device_numbers = Some(format!("serial:{serial:#010x}")); + } + let device_ids = rustfs_utils::os::get_physical_device_ids(&canonical_path).map_err(|err| { + Error::other(format!("failed to inspect physical disk for local endpoint '{canonical_path}': {err}")) + })?; + diagnostic.device_ids = Some(device_ids.clone()); + diagnostics.push(diagnostic); + + for device_id in device_ids { + device_paths.entry(device_id).or_default().insert(canonical_path.clone()); + } + } + + #[cfg(not(windows))] + if !missing_paths.is_empty() { + warn!( + missing_paths = ?missing_paths, + "Excluding non-existent local endpoint paths from physical disk independence validation during endpoint parsing", + ); + } + + warn!( + diagnostics = %diagnostics + .iter() + .map(LocalDiskValidationDiagnostic::summary) + .collect::>() + .join("; "), + "Collected local endpoint disk-topology diagnostics before physical disk independence validation", + ); + + let shared_devices = device_paths + .into_iter() + .filter_map(|(device_id, paths)| { + if paths.len() <= 1 { + return None; + } + + Some((device_id, paths.into_iter().collect::>())) + }) + .collect::>(); + + if shared_devices.is_empty() { + return Ok(()); + } + + let details = shared_devices + .into_iter() + .map(|(device_id, paths)| format!("{device_id} => {}", paths.join(", "))) + .collect::>() + .join("; "); + let diagnostics_summary = diagnostics + .iter() + .map(LocalDiskValidationDiagnostic::summary) + .collect::>() + .join("; "); + + Err(Error::other(format!( + "local erasure endpoints must use distinct physical disks; detected shared devices [{details}]. \ +validation diagnostics: [{diagnostics_summary}]. \ +Set {ENV_UNSAFE_BYPASS_DISK_CHECK}=true only for local testing or CI to bypass this safety check" + ))) +} + +fn validate_local_cross_device_mounts(local_paths: &[String]) -> Result<()> { + rustfs_utils::os::check_cross_device_mounts(local_paths) + .map_err(|err| Error::other(format!("local endpoint cross-device mount validation failed: {err}"))) +} + +#[cfg(test)] +mod test { + use path_absolutize::Absolutize; + use rustfs_utils::must_get_local_ips; + + use super::*; + + #[cfg(target_os = "linux")] + use serial_test::serial; + use std::path::Path; + #[cfg(target_os = "linux")] + use temp_env::async_with_vars; + #[cfg(target_os = "linux")] + use tempfile::tempdir; + + #[test] + fn test_new_endpoints() { + let test_cases = [ + (vec!["/d1", "/d2", "/d3", "/d4"], None, 1), + ( + vec![ + "http://localhost/d1", + "http://localhost/d2", + "http://localhost/d3", + "http://localhost/d4", + ], + None, + 2, + ), + ( + vec![ + "http://example.org/d1", + "http://example.com/d1", + "http://example.net/d1", + "http://example.edu/d1", + ], + None, + 3, + ), + ( + vec![ + "http://localhost/d1", + "http://localhost/d2", + "http://example.org/d1", + "http://example.org/d2", + ], + None, + 4, + ), + ( + vec![ + "https://localhost:9000/d1", + "https://localhost:9001/d2", + "https://localhost:9002/d3", + "https://localhost:9003/d4", + ], + None, + 5, + ), + // It is valid WRT endpoint list that same path is expected with different port on same server. + ( + vec![ + "https://127.0.0.1:9000/d1", + "https://127.0.0.1:9001/d1", + "https://127.0.0.1:9002/d1", + "https://127.0.0.1:9003/d1", + ], + None, + 6, + ), + (vec!["d1", "d2", "d3", "d1"], Some(Error::other("duplicate endpoints found")), 7), + (vec!["d1", "d2", "d3", "./d1"], Some(Error::other("duplicate endpoints found")), 8), + ( + vec![ + "http://localhost/d1", + "http://localhost/d2", + "http://localhost/d1", + "http://localhost/d4", + ], + Some(Error::other("duplicate endpoints found")), + 9, + ), + ( + vec!["ftp://server/d1", "http://server/d2", "http://server/d3", "http://server/d4"], + Some(Error::other("'ftp://server/d1': io error invalid URL endpoint format")), + 10, + ), + ( + vec!["d1", "http://localhost/d2", "d3", "d4"], + Some(Error::other("mixed style endpoints are not supported")), + 11, + ), + ( + vec![ + "http://example.org/d1", + "https://example.com/d1", + "http://example.net/d1", + "https://example.edut/d1", + ], + Some(Error::other("mixed scheme is not supported")), + 12, + ), + ( + vec![ + "192.168.1.210:9000/tmp/dir0", + "192.168.1.210:9000/tmp/dir1", + "192.168.1.210:9000/tmp/dir2", + "192.168.110:9000/tmp/dir3", + ], + Some(Error::other("'192.168.1.210:9000/tmp/dir0': io error")), + 13, + ), + ]; + + for test_case in test_cases { + let args: Vec = test_case.0.iter().map(|v| v.to_string()).collect(); + let ret = Endpoints::try_from(args.as_slice()); + + match (test_case.1, ret) { + (None, Err(e)) => panic!("{}: error: expected = , got = {}", test_case.2, e), + (None, Ok(_)) => {} + (Some(e), Ok(_)) => panic!("{}: error: expected = {}, got = ", test_case.2, e), + (Some(e), Err(e2)) => { + assert!( + e2.to_string().starts_with(&e.to_string()), + "{}: error: expected = {}, got = {}", + test_case.2, + e, + e2 + ) + } + } + } + } + + #[tokio::test] + async fn test_create_pool_endpoints() { + #[derive(Default)] + struct TestCase<'a> { + num: usize, + server_addr: &'a str, + args: Vec<&'a str>, + expected_endpoints: Option, + expected_setup_type: Option, + expected_err: Option, + } + + // Filter ipList by IPs those do not start with '127.'. + let non_loop_back_i_ps = + must_get_local_ips().map_or(vec![], |v| v.into_iter().filter(|ip| ip.is_ipv4() && ip.is_loopback()).collect()); + if non_loop_back_i_ps.is_empty() { + panic!("No non-loop back IP address found for this host"); + } + let non_loop_back_ip = non_loop_back_i_ps[0]; + let remote_ip1 = "192.0.2.10"; + let remote_ip2 = "192.0.2.11"; + let remote_ip3 = "192.0.2.12"; + + let case1_endpoint1 = format!("http://{non_loop_back_ip}/d1"); + let case1_endpoint2 = format!("http://{non_loop_back_ip}/d2"); + let args = vec![ + format!("http://{}:10000/d1", non_loop_back_ip), + format!("http://{}:10000/d2", non_loop_back_ip), + format!("http://{remote_ip1}:10000/d3"), + format!("http://{remote_ip2}:10000/d4"), + ]; + let (case1_ur_ls, case1_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:10000/")); + + let case2_endpoint1 = format!("http://{non_loop_back_ip}/d1"); + let case2_endpoint2 = format!("http://{non_loop_back_ip}:9000/d2"); + let args = vec![ + format!("http://{}:10000/d1", non_loop_back_ip), + format!("http://{}:9000/d2", non_loop_back_ip), + format!("http://{remote_ip1}:10000/d3"), + format!("http://{remote_ip2}:10000/d4"), + ]; + let (case2_ur_ls, case2_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:10000/")); + + let case3_endpoint1 = format!("http://{non_loop_back_ip}/d1"); + let args = vec![ + format!("http://{}:80/d1", non_loop_back_ip), + format!("http://{remote_ip1}:9000/d2"), + format!("http://{remote_ip2}:80/d3"), + format!("http://{remote_ip3}:80/d4"), + ]; + let (case3_ur_ls, case3_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:80/")); + + let case4_endpoint1 = format!("http://{non_loop_back_ip}/d1"); + let args = vec![ + format!("http://{}:9000/d1", non_loop_back_ip), + format!("http://{remote_ip1}:9000/d2"), + format!("http://{remote_ip2}:9000/d3"), + format!("http://{remote_ip3}:9000/d4"), + ]; + let (case4_ur_ls, case4_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9000/")); + + let case5_endpoint1 = format!("http://{non_loop_back_ip}:9000/d1"); + let case5_endpoint2 = format!("http://{non_loop_back_ip}:9001/d2"); + let case5_endpoint3 = format!("http://{non_loop_back_ip}:9002/d3"); + let case5_endpoint4 = format!("http://{non_loop_back_ip}:9003/d4"); + let args = vec![ + case5_endpoint1.clone(), + case5_endpoint2.clone(), + case5_endpoint3.clone(), + case5_endpoint4.clone(), + ]; + let (case5_ur_ls, case5_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9000/")); + + let case6_endpoint1 = format!("http://{non_loop_back_ip}:9003/d4"); + let args = vec![ + "http://127.0.0.1:9000/d1".to_string(), + "http://127.0.0.1:9001/d2".to_string(), + "http://127.0.0.1:9002/d3".to_string(), + case6_endpoint1.clone(), + ]; + let (case6_ur_ls, case6_local_flags) = get_expected_endpoints(args, format!("http://{non_loop_back_ip}:9003/")); + + let case7_endpoint1 = format!("http://{non_loop_back_ip}:9001/export"); + let case7_endpoint2 = format!("http://{non_loop_back_ip}:9000/export"); + + let test_cases = [ + TestCase { + num: 1, + server_addr: "localhost", + expected_err: Some(Error::other("address localhost: missing port in address")), + ..Default::default() + }, + // Erasure Single Drive + TestCase { + num: 2, + server_addr: "127.0.0.1:9000", + args: vec!["http://127.0.0.1/d1"], + expected_err: Some(Error::other("use path style endpoint for single node setup")), + ..Default::default() + }, + TestCase { + num: 3, + server_addr: "0.0.0.0:443", + args: vec!["/d1"], + expected_endpoints: Some(Endpoints(vec![Endpoint { + url: must_file_path("/d1"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }])), + expected_setup_type: Some(SetupType::ErasureSD), + ..Default::default() + }, + TestCase { + num: 4, + server_addr: "127.0.0.1:10000", + args: vec!["/d1"], + expected_endpoints: Some(Endpoints(vec![Endpoint { + url: must_file_path("/d1"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }])), + expected_setup_type: Some(SetupType::ErasureSD), + ..Default::default() + }, + TestCase { + num: 5, + server_addr: "127.0.0.1:9000", + args: vec![ + "https://127.0.0.1:9000/d1", + "https://127.0.0.1:9001/d1", + "https://192.0.2.1/d1", + "https://192.0.2.1/d2", + ], + expected_err: Some(Error::other("same path '/d1' can not be served by different port on same address")), + ..Default::default() + }, + // Erasure Setup with PathEndpointType + TestCase { + num: 6, + server_addr: "0.0.0.0:1234", + args: vec!["/d1", "/d2", "/d3", "/d4"], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: must_file_path("/d1"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: must_file_path("/d2"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: must_file_path("/d3"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: must_file_path("/d4"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::Erasure), + ..Default::default() + }, + // DistErasure Setup with URLEndpointType + TestCase { + num: 7, + server_addr: "0.0.0.0:9000", + args: vec![ + "http://127.0.0.1/d1", + "http://127.0.0.1/d2", + "http://127.0.0.1/d3", + "http://127.0.0.1/d4", + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: must_url("http://127.0.0.1:9000/d1"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: must_url("http://127.0.0.1:9000/d2"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: must_url("http://127.0.0.1:9000/d3"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: must_url("http://127.0.0.1:9000/d4"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::Erasure), + ..Default::default() + }, + // DistErasure Setup with URLEndpointType having mixed naming to local host. + TestCase { + num: 8, + server_addr: "127.0.0.1:10000", + args: vec![ + "http://[::1]/d1", + "http://[::1]/d2", + "http://127.0.0.1/d3", + "http://127.0.0.1/d4", + ], + expected_err: Some(Error::other("all local endpoints should not have different hostnames/ips")), + ..Default::default() + }, + TestCase { + num: 9, + server_addr: "0.0.0.0:9001", + args: vec![ + "http://10.0.0.1:9000/export", + "http://10.0.0.2:9000/export", + case7_endpoint1.as_str(), + "http://10.0.0.2:9001/export", + ], + expected_err: Some(Error::other("same path '/export' can not be served by different port on same address")), + ..Default::default() + }, + TestCase { + num: 10, + server_addr: "0.0.0.0:9000", + args: vec![ + "http://127.0.0.1:9000/export", + case7_endpoint2.as_str(), + "http://10.0.0.1:9000/export", + "http://10.0.0.2:9000/export", + ], + expected_err: Some(Error::other("path '/export' cannot be served by different address on same server")), + ..Default::default() + }, + // DistErasure type + TestCase { + num: 11, + server_addr: "127.0.0.1:10000", + args: vec![ + case1_endpoint1.as_str(), + case1_endpoint2.as_str(), + "http://192.0.2.10/d3", + "http://192.0.2.11/d4", + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: case1_ur_ls[0].clone(), + is_local: case1_local_flags[0], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case1_ur_ls[1].clone(), + is_local: case1_local_flags[1], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case1_ur_ls[2].clone(), + is_local: case1_local_flags[2], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case1_ur_ls[3].clone(), + is_local: case1_local_flags[3], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::DistErasure), + ..Default::default() + }, + TestCase { + num: 12, + server_addr: "127.0.0.1:10000", + args: vec![ + case2_endpoint1.as_str(), + case2_endpoint2.as_str(), + "http://192.0.2.10/d3", + "http://192.0.2.11/d4", + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: case2_ur_ls[0].clone(), + is_local: case2_local_flags[0], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case2_ur_ls[1].clone(), + is_local: case2_local_flags[1], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case2_ur_ls[2].clone(), + is_local: case2_local_flags[2], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case2_ur_ls[3].clone(), + is_local: case2_local_flags[3], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::DistErasure), + ..Default::default() + }, + TestCase { + num: 13, + server_addr: "0.0.0.0:80", + args: vec![ + case3_endpoint1.as_str(), + "http://192.0.2.10:9000/d2", + "http://192.0.2.11/d3", + "http://192.0.2.12/d4", + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: case3_ur_ls[0].clone(), + is_local: case3_local_flags[0], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case3_ur_ls[1].clone(), + is_local: case3_local_flags[1], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case3_ur_ls[2].clone(), + is_local: case3_local_flags[2], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case3_ur_ls[3].clone(), + is_local: case3_local_flags[3], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::DistErasure), + ..Default::default() + }, + TestCase { + num: 14, + server_addr: "0.0.0.0:9000", + args: vec![ + case4_endpoint1.as_str(), + "http://192.0.2.10/d2", + "http://192.0.2.11/d3", + "http://192.0.2.12/d4", + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: case4_ur_ls[0].clone(), + is_local: case4_local_flags[0], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case4_ur_ls[1].clone(), + is_local: case4_local_flags[1], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case4_ur_ls[2].clone(), + is_local: case4_local_flags[2], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case4_ur_ls[3].clone(), + is_local: case4_local_flags[3], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::DistErasure), + ..Default::default() + }, + TestCase { + num: 15, + server_addr: "0.0.0.0:9000", + args: vec![ + case5_endpoint1.as_str(), + case5_endpoint2.as_str(), + case5_endpoint3.as_str(), + case5_endpoint4.as_str(), + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: case5_ur_ls[0].clone(), + is_local: case5_local_flags[0], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case5_ur_ls[1].clone(), + is_local: case5_local_flags[1], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case5_ur_ls[2].clone(), + is_local: case5_local_flags[2], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case5_ur_ls[3].clone(), + is_local: case5_local_flags[3], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::DistErasure), + ..Default::default() + }, + TestCase { + num: 16, + server_addr: "0.0.0.0:9003", + args: vec![ + "http://127.0.0.1:9000/d1", + "http://127.0.0.1:9001/d2", + "http://127.0.0.1:9002/d3", + case6_endpoint1.as_str(), + ], + expected_endpoints: Some(Endpoints(vec![ + Endpoint { + url: case6_ur_ls[0].clone(), + is_local: case6_local_flags[0], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case6_ur_ls[1].clone(), + is_local: case6_local_flags[1], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case6_ur_ls[2].clone(), + is_local: case6_local_flags[2], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: case6_ur_ls[3].clone(), + is_local: case6_local_flags[3], + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + ])), + expected_setup_type: Some(SetupType::DistErasure), + ..Default::default() + }, + ]; + + for test_case in test_cases { + let disks_layout = match DisksLayout::from_volumes(test_case.args.as_slice()) { + Ok(v) => v, + Err(e) => { + if test_case.expected_err.is_none() { + panic!("Test {}: unexpected error: {}", test_case.num, e); + } + continue; + } + }; + + match ( + test_case.expected_err, + PoolEndpointList::create_pool_endpoints(test_case.server_addr, &disks_layout).await, + ) { + (None, Err(err)) => panic!("Test {}: error: expected = , got = {}", test_case.num, err), + (Some(err), Ok(_)) => panic!("Test {}: error: expected = {}, got = ", test_case.num, err), + (Some(e), Err(e2)) => { + assert_eq!( + e.to_string(), + e2.to_string(), + "Test {}: error: expected = {}, got = {}", + test_case.num, + e, + e2 + ) + } + (None, Ok(pools)) => { + if Some(&pools.setup_type) != test_case.expected_setup_type.as_ref() { + panic!( + "Test {}: setupType: expected = {:?}, got = {:?}", + test_case.num, test_case.expected_setup_type, &pools.setup_type + ) + } + + let left_len = test_case.expected_endpoints.as_ref().map(|v| v.as_ref().len()); + let right_len = pools.as_ref().first().map(|v| v.as_ref().len()); + + if left_len != right_len { + panic!("Test {}: endpoints len: expected = {:?}, got = {:?}", test_case.num, left_len, right_len); + } + + for (i, ep) in pools.as_ref()[0].as_ref().iter().enumerate() { + assert_eq!( + ep.to_string(), + test_case.expected_endpoints.as_ref().unwrap().as_ref()[i].to_string(), + "Test {}: endpoints: expected = {}, got = {}", + test_case.num, + test_case.expected_endpoints.as_ref().unwrap().as_ref()[i], + ep + ) + } + } + } + } + } + + fn must_file_path(s: impl AsRef) -> url::Url { + let path = s.as_ref().absolutize().expect("absolute test path"); + let url = url::Url::from_file_path(&path); + + assert!(url.is_ok(), "failed to convert path to URL: {}", path.display()); + + url.unwrap() + } + + fn must_url(s: &str) -> url::Url { + url::Url::parse(s).unwrap() + } + + fn get_expected_endpoints(args: Vec, prefix: String) -> (Vec, Vec) { + let mut urls = vec![]; + let mut local_flags = vec![]; + for arg in args { + urls.push(url::Url::parse(&arg).unwrap()); + local_flags.push(arg.starts_with(&prefix)); + } + + (urls, local_flags) + } + + #[tokio::test] + async fn test_create_server_endpoints() { + let test_cases = [ + // Invalid input. + ("", vec![], false), + // Range cannot be negative. + ("0.0.0.0:9000", vec!["/export1{-1...1}"], false), + // Range cannot start bigger than end. + ("0.0.0.0:9000", vec!["/export1{64...1}"], false), + // Range can only be numeric. + ("0.0.0.0:9000", vec!["/export1{a...z}"], false), + // Duplicate disks not allowed. + ("0.0.0.0:9000", vec!["/export1{1...32}", "/export1{1...32}"], false), + // Same host cannot export same disk on two ports - special case localhost. + ("0.0.0.0:9001", vec!["http://localhost:900{1...2}/export{1...64}"], false), + // Valid inputs. + ("0.0.0.0:9000", vec!["/export1"], true), + ("0.0.0.0:9000", vec!["/export1", "/export2", "/export3", "/export4"], true), + ("0.0.0.0:9000", vec!["/export1{1...64}"], true), + ("0.0.0.0:9000", vec!["/export1{01...64}"], true), + ("0.0.0.0:9000", vec!["/export1{1...32}", "/export1{33...64}"], true), + ("0.0.0.0:9001", vec!["http://localhost:9001/export{1...64}"], true), + ("0.0.0.0:9001", vec!["http://localhost:9001/export{01...64}"], true), + ]; + + for (i, test_case) in test_cases.iter().enumerate() { + let disks_layout = match DisksLayout::from_volumes(test_case.1.as_slice()) { + Ok(v) => v, + Err(e) => { + if test_case.2 { + panic!("Test {}: unexpected error: {}", i + 1, e); + } + continue; + } + }; + + let ret = EndpointServerPools::create_server_endpoints(test_case.0, &disks_layout).await; + + if let Err(err) = ret { + if test_case.2 { + panic!("Test {}: Expected success but failed instead {}", i + 1, err) + } + } else if !test_case.2 { + panic!("Test {}: expected failure but passed instead", i + 1); + } + } + } + + #[cfg(target_os = "linux")] + #[serial] + #[tokio::test] + async fn reject_shared_local_physical_disks_by_default() { + async_with_vars([(ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>), (ENV_MINIO_CI, None::<&str>)], async { + let dir = tempdir().unwrap(); + let disk1 = dir.path().join("disk1"); + let disk2 = dir.path().join("disk2"); + std::fs::create_dir_all(&disk1).unwrap(); + std::fs::create_dir_all(&disk2).unwrap(); + + let args = vec![disk1.to_string_lossy().into_owned(), disk2.to_string_lossy().into_owned()]; + let layout = DisksLayout::from_volumes(args.as_slice()).unwrap(); + + let err = EndpointServerPools::create_server_endpoints("0.0.0.0:9000", &layout) + .await + .unwrap_err(); + + let err_text = err.to_string(); + assert!(err_text.contains("distinct physical disks"), "unexpected error: {err_text}"); + assert!(err_text.contains(ENV_UNSAFE_BYPASS_DISK_CHECK), "unexpected error: {err_text}"); + assert!(err_text.contains("validation diagnostics:"), "unexpected error: {err_text}"); + assert!(err_text.contains("st_dev='"), "unexpected error: {err_text}"); + assert!(err_text.contains("device_ids=["), "unexpected error: {err_text}"); + }) + .await; + } + + #[cfg(target_os = "linux")] + #[serial] + #[tokio::test] + async fn allow_shared_local_physical_disks_with_explicit_env_bypass() { + async_with_vars([(ENV_UNSAFE_BYPASS_DISK_CHECK, Some("true"))], async { + let dir = tempdir().unwrap(); + let disk1 = dir.path().join("disk1"); + let disk2 = dir.path().join("disk2"); + std::fs::create_dir_all(&disk1).unwrap(); + std::fs::create_dir_all(&disk2).unwrap(); + + let args = vec![disk1.to_string_lossy().into_owned(), disk2.to_string_lossy().into_owned()]; + let layout = DisksLayout::from_volumes(args.as_slice()).unwrap(); + + let ret = EndpointServerPools::create_server_endpoints("0.0.0.0:9000", &layout).await; + assert!(ret.is_ok(), "expected bypassed disk validation to succeed, got {ret:?}"); + }) + .await; + } + + #[cfg(target_os = "linux")] + #[serial] + #[tokio::test] + async fn allow_shared_local_physical_disks_with_minio_ci_alias() { + async_with_vars([(ENV_UNSAFE_BYPASS_DISK_CHECK, None::<&str>), (ENV_MINIO_CI, Some("1"))], async { + let dir = tempdir().unwrap(); + let disk1 = dir.path().join("disk1"); + let disk2 = dir.path().join("disk2"); + std::fs::create_dir_all(&disk1).unwrap(); + std::fs::create_dir_all(&disk2).unwrap(); + + let args = vec![disk1.to_string_lossy().into_owned(), disk2.to_string_lossy().into_owned()]; + let layout = DisksLayout::from_volumes(args.as_slice()).unwrap(); + + let ret = EndpointServerPools::create_server_endpoints("0.0.0.0:9000", &layout).await; + assert!(ret.is_ok(), "expected MINIO_CI alias to bypass disk validation, got {ret:?}"); + }) + .await; + } +} diff --git a/crates/ecstore/src/layout/mod.rs b/crates/ecstore/src/layout/mod.rs index 0fd12aefc..959bf2ae6 100644 --- a/crates/ecstore/src/layout/mod.rs +++ b/crates/ecstore/src/layout/mod.rs @@ -5,5 +5,7 @@ //! file moves happen. pub(crate) mod disks_layout; +pub(crate) mod endpoint; +pub(crate) mod endpoints; pub(crate) mod format; pub(crate) mod set_layout; diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 70d9e4f75..5f4c47bd6 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,17 +5,16 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-ecstore-layout-format-move` -- Baseline: `overtrue/arch-ecstore-layout-foundation` - (`2e10661d088a665c789fdb092871c6b52675aa1e`). -- Stacked on: local E-001/E-SET-001, which is stacked on local R-033 and - local R-032. +- Branch: `overtrue/arch-ecstore-layout-endpoints-move` +- Baseline: merged `E-002/E-LAYOUT-001`. +- Stacked on: merged ECStore layout foundation and format layout ownership + slices. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: pure-move ECStore format and disk-layout expansion modules - into the internal layout bucket while preserving old public paths. +- Rust code changes: pure-move ECStore endpoint parsing and endpoint grouping + modules into the internal layout bucket while preserving old public paths. - CI/script changes: none. -- Docs changes: record the ECStore format/disk-layout pure move slice. +- Docs changes: record the ECStore endpoint pure move slice. ## Phase 0 Tasks @@ -2241,10 +2240,23 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block diff hygiene, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `E-003/E-LAYOUT-002` Move ECStore endpoint layout owners. + - Do: pure-move endpoint parsing and endpoint grouping into the ECStore + layout bucket while keeping compatibility stubs at the old public paths. + - Acceptance: `crate::disk::endpoint::*` and `crate::endpoints::*` remain + usable, `layout::endpoint` owns `Endpoint`, and `layout::endpoints` owns + `EndpointServerPools` and endpoint grouping. + - Must preserve: endpoint string parsing, URL/path validation, local-host + detection, pool/set/disk indexes, endpoint grouping, disk independence + checks, setup type classification, and old public module paths. + - Verification: focused ECStore endpoint tests, ECStore/RustFS/Heal compile + checks, migration/layer guards, formatting, diff hygiene, Rust risk scan, + branch freshness check, pre-commit quality gate, and three-expert review. + ## Next PRs -1. `pure-move`: continue moving endpoint grouping and runtime-neutral layout - helpers once E-002/E-LAYOUT-001 lands. +1. `pure-move`: continue moving runtime-neutral pool/set layout helpers once + E-003/E-LAYOUT-002 lands. 2. `pure-move`: continue pruning residual embedded startup-only orchestration once the lifecycle helpers are merged. @@ -2252,9 +2264,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | E-002/E-LAYOUT-001 is a pure move into ECStore layout with compatibility stubs at old public paths; no new runtime owner or dependency boundary is introduced. | -| Migration preservation | passed | Format JSON, disk UUID lookup, distribution algorithm, disk-layout expansion, and old public module paths remain preserved. | -| Testing/verification | passed | Focused format/layout checks, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | +| Quality/architecture | passed | E-003/E-LAYOUT-002 is a pure move into ECStore layout with compatibility stubs at old public paths; no new runtime owner or dependency boundary is introduced. | +| Migration preservation | passed | Endpoint parsing, local-host detection, pool/set/disk indexes, endpoint grouping, disk independence checks, setup type classification, and old public module paths remain preserved. | +| Testing/verification | passed | Focused endpoint/layout checks, compile checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | ## Verification Notes @@ -2329,6 +2341,19 @@ Passed before push: - `make pre-commit`: passed. - Three-expert review: passed. +- Issue #660 E-003/E-LAYOUT-002 current slice: + - `cargo test -p rustfs-ecstore layout::endpoint -- --nocapture`: passed. + - `cargo test -p rustfs-ecstore layout::endpoints -- --nocapture`: passed. + - `cargo check -p rustfs-ecstore -p rustfs -p rustfs-heal`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `cargo fmt --all --check`: passed. + - `git diff --check`: passed. + - Rust risk scan on changed Rust files: passed; only existing endpoint + production/test unwrap and expectation paths were moved. + - `make pre-commit`: passed. + - Three-expert review: passed. + - Issue #660 X-012 current slice: - `cargo test -p rustfs-extension-schema`: passed. - `cargo check -p rustfs-extension-schema`: passed.