Merge branch 'endpoint' into erasure

This commit is contained in:
shiro.lee
2024-07-06 14:01:14 +08:00
8 changed files with 1314 additions and 111 deletions
+12 -5
View File
@@ -37,7 +37,7 @@ impl PoolDisksLayout {
self.layout.len()
}
pub fn as_cmd_line(&self) -> &str {
pub fn get_cmd_line(&self) -> &str {
&self.cmd_line
}
}
@@ -283,11 +283,11 @@ fn possible_set_counts_with_symmetry(set_counts: &[usize], arg_patterns: &[ArgPa
for &ss in set_counts {
let mut symmetry = false;
for arg_pattern in arg_patterns {
for p in arg_pattern.inner.iter() {
if p.seq.len() > ss {
symmetry = (p.seq.len() % ss) == 0;
for p in arg_pattern.as_ref().iter() {
if p.len() > ss {
symmetry = (p.len() % ss) == 0;
} else {
symmetry = (ss % p.seq.len()) == 0;
symmetry = (ss % p.len()) == 0;
}
}
}
@@ -520,6 +520,13 @@ mod test {
]],
success: true,
},
TestCase {
num: 15,
args: vec!["https://node{1...3}.example.net/mnt/drive{1...8}"],
total_sizes: vec![24],
indexes: vec![vec![12, 12]],
success: true,
},
];
for test_case in test_cases {
+20 -4
View File
@@ -16,9 +16,9 @@ const ELLIPSES: &str = "...";
/// associated prefix and suffixes.
#[derive(Debug, Default, PartialEq, Eq)]
pub struct Pattern {
pub prefix: String,
pub suffix: String,
pub seq: Vec<String>,
pub(crate) prefix: String,
pub(crate) suffix: String,
pub(crate) seq: Vec<String>,
}
impl Pattern {
@@ -37,12 +37,28 @@ impl Pattern {
ret
}
pub fn len(&self) -> usize {
self.seq.len()
}
}
/// contains a list of patterns provided in the input.
#[derive(Debug, PartialEq, Eq)]
pub struct ArgPattern {
pub inner: Vec<Pattern>,
inner: Vec<Pattern>,
}
impl AsRef<Vec<Pattern>> for ArgPattern {
fn as_ref(&self) -> &Vec<Pattern> {
&self.inner
}
}
impl AsMut<Vec<Pattern>> for ArgPattern {
fn as_mut(&mut self) -> &mut Vec<Pattern> {
&mut self.inner
}
}
impl ArgPattern {
+1190 -63
View File
File diff suppressed because it is too large Load Diff
+9 -9
View File
@@ -14,7 +14,7 @@ type Client = Arc<Box<dyn PeerS3Client>>;
#[async_trait]
pub trait PeerS3Client: Debug + Sync + Send + 'static {
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
fn get_pools(&self) -> Vec<i32>;
fn get_pools(&self) -> Vec<usize>;
}
#[derive(Debug)]
@@ -53,7 +53,7 @@ impl S3PeerSys {
#[async_trait]
impl PeerS3Client for S3PeerSys {
fn get_pools(&self) -> Vec<i32> {
fn get_pools(&self) -> Vec<usize> {
unimplemented!()
}
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
@@ -80,7 +80,7 @@ impl PeerS3Client for S3PeerSys {
let mut per_pool_errs = Vec::with_capacity(self.clients.len());
for (j, cli) in self.clients.iter().enumerate() {
let pools = cli.get_pools();
let idx = i as i32;
let idx = i;
if pools.contains(&idx) {
per_pool_errs.push(errors[j].as_ref());
}
@@ -99,11 +99,11 @@ impl PeerS3Client for S3PeerSys {
pub struct LocalPeerS3Client {
pub local_disks: Vec<DiskStore>,
pub node: Node,
pub pools: Vec<i32>,
pub pools: Vec<usize>,
}
impl LocalPeerS3Client {
fn new(local_disks: Vec<DiskStore>, node: Node, pools: Vec<i32>) -> Self {
fn new(local_disks: Vec<DiskStore>, node: Node, pools: Vec<usize>) -> Self {
Self {
local_disks,
node,
@@ -114,7 +114,7 @@ impl LocalPeerS3Client {
#[async_trait]
impl PeerS3Client for LocalPeerS3Client {
fn get_pools(&self) -> Vec<i32> {
fn get_pools(&self) -> Vec<usize> {
self.pools.clone()
}
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
@@ -154,18 +154,18 @@ impl PeerS3Client for LocalPeerS3Client {
#[derive(Debug)]
pub struct RemotePeerS3Client {
pub node: Node,
pub pools: Vec<i32>,
pub pools: Vec<usize>,
}
impl RemotePeerS3Client {
fn new(node: Node, pools: Vec<i32>) -> Self {
fn new(node: Node, pools: Vec<usize>) -> Self {
Self { node, pools }
}
}
#[async_trait]
impl PeerS3Client for RemotePeerS3Client {
fn get_pools(&self) -> Vec<i32> {
fn get_pools(&self) -> Vec<usize> {
unimplemented!()
}
async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> {
+7 -6
View File
@@ -29,20 +29,21 @@ pub struct ECStore {
impl ECStore {
pub async fn new(address: String, endpoints: Vec<String>) -> Result<Self> {
let layouts = DisksLayout::try_from(endpoints.as_slice())?;
let layouts = DisksLayout::try_from(endpoints.as_slice()).map_err(|v| Error::msg(v))?;
let mut deployment_id = None;
let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?;
let (endpoint_pools, _) =
EndpointServerPools::create_server_endpoints(address.as_str(), &layouts).map_err(|v| Error::msg(v))?;
let mut pools = Vec::with_capacity(endpoint_pools.len());
let mut disk_map = HashMap::with_capacity(endpoint_pools.len());
let mut pools = Vec::with_capacity(endpoint_pools.as_ref().len());
let mut disk_map = HashMap::with_capacity(endpoint_pools.as_ref().len());
let first_is_local = endpoint_pools.first_is_local();
let first_is_local = endpoint_pools.first_local();
let mut local_disks = Vec::new();
for (i, pool_eps) in endpoint_pools.iter().enumerate() {
for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() {
// TODO: read from config parseStorageClass
let partiy_count = store_init::default_partiy_count(pool_eps.drives_per_set);
+66 -21
View File
@@ -1,13 +1,20 @@
use crate::error::{Error, Result};
use lazy_static::lazy_static;
use std::{
collections::HashSet,
net::{IpAddr, SocketAddr, ToSocketAddrs},
};
use url::Host;
lazy_static! {
static ref LOCAL_IPS: Vec<IpAddr> = must_get_local_ips().unwrap();
}
/// helper for validating if the provided arg is an ip address.
pub fn is_socket_addr(host: &str) -> bool {
host.parse::<SocketAddr>().is_ok() || host.parse::<IpAddr>().is_ok()
pub fn is_socket_addr(addr: &str) -> bool {
// TODO IPv6 zone information?
addr.parse::<SocketAddr>().is_ok() || addr.parse::<IpAddr>().is_ok()
}
/// checks if server_addr is valid and local host.
@@ -30,30 +37,18 @@ pub fn check_local_server_addr(server_addr: &str) -> Result<SocketAddr> {
SocketAddr::V6(a) => Host::Ipv6(*a.ip()),
};
if is_local_host(host, 0, 0).is_ok() {
if is_local_host(host, 0, 0)? {
return Ok(a);
}
}
return Err(Error::from_string("host in server address should be this server"));
}
pub fn split_host_port(s: &str) -> Result<(String, u16)> {
let parts: Vec<&str> = s.split(':').collect();
if parts.len() == 2 {
if let Ok(port) = parts[1].parse::<u16>() {
return Ok((parts[0].to_string(), port));
}
}
Err(Error::from_string("Invalid address format or port number"))
Err(Error::from_string("host in server address should be this server"))
}
/// checks if the given parameter correspond to one of
/// the local IP of the current machine
pub fn is_local_host(host: Host<&str>, port: u16, local_port: u16) -> Result<bool> {
let local_ips = must_get_local_ips();
let local_set: HashSet<IpAddr> = local_ips.into_iter().collect();
let local_set: HashSet<IpAddr> = LOCAL_IPS.iter().copied().collect();
let is_local_host = match host {
Host::Domain(domain) => {
let ips = match (domain, 0).to_socket_addrs().map(|v| v.map(|v| v.ip()).collect::<Vec<_>>()) {
@@ -74,11 +69,34 @@ pub fn is_local_host(host: Host<&str>, port: u16, local_port: u16) -> Result<boo
Ok(is_local_host)
}
/// returns IP address of given host.
pub fn get_host_ip(host: Host<&str>) -> Result<HashSet<IpAddr>> {
match host {
Host::Domain(domain) => match (domain, 0)
.to_socket_addrs()
.map(|v| v.map(|v| v.ip()).collect::<HashSet<_>>())
{
Ok(ips) => Ok(ips),
Err(err) => Err(Error::new(Box::new(err))),
},
Host::Ipv4(ip) => {
let mut set = HashSet::with_capacity(1);
set.insert(IpAddr::V4(ip));
Ok(set)
}
Host::Ipv6(ip) => {
let mut set = HashSet::with_capacity(1);
set.insert(IpAddr::V6(ip));
Ok(set)
}
}
}
/// returns IPs of local interface
fn must_get_local_ips() -> Vec<IpAddr> {
pub(crate) fn must_get_local_ips() -> Result<Vec<IpAddr>> {
match netif::up() {
Ok(up) => up.map(|x| x.address().to_owned()).collect(),
Err(_) => vec![],
Ok(up) => Ok(up.map(|x| x.address().to_owned()).collect()),
Err(err) => Err(Error::from_string(format!("Unable to get IP addresses of this host: {}", err))),
}
}
@@ -106,9 +124,36 @@ mod test {
}
}
#[test]
fn test_check_local_server_addr() {
let test_cases = [
(":54321", Ok(())),
("localhost:54321", Ok(())),
("0.0.0.0:9000", Ok(())),
(":0", Ok(())),
("localhost", Err(Error::from_string("invalid socket address"))),
("", Err(Error::from_string("invalid socket address"))),
(
"example.org:54321",
Err(Error::from_string("host in server address should be this server")),
),
(":-10", Err(Error::from_string("invalid port value"))),
];
for test_case in test_cases {
let ret = check_local_server_addr(test_case.0);
if test_case.1.is_ok() && ret.is_err() {
panic!("{}: error: expected = <nil>, got = {:?}", test_case.0, ret);
}
if test_case.1.is_err() && ret.is_ok() {
panic!("{}: error: expected = {:?}, got = <nil>", test_case.0, test_case.1);
}
}
}
#[test]
fn test_must_get_local_ips() {
let local_ips = must_get_local_ips();
let local_ips = must_get_local_ips().unwrap();
let local_set: HashSet<IpAddr> = local_ips.into_iter().collect();
assert!(local_set.contains(&IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1))));