diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index 606a1df61..acf781fe4 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -4,11 +4,11 @@ use std::{ }; use anyhow::Error; -use bytes::Bytes; use futures::future::join_all; use path_absolutize::Absolutize; use time::OffsetDateTime; -use tokio::fs; +use tokio::fs::{self, File}; +use tokio::io::ErrorKind; use uuid::Uuid; use crate::{ @@ -17,11 +17,12 @@ use crate::{ }; pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys"; +pub const BUCKET_META_PREFIX: &str = "buckets"; pub const FORMAT_CONFIG_FILE: &str = "format.json"; pub struct DiskOption { pub cleanup: bool, - pub health_check: bool, + pub _health_check: bool, } pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result { @@ -120,9 +121,26 @@ impl LocalDisk { format_last_check, }; + disk.make_meta_volumes().await?; + Ok(disk) } + async fn make_meta_volumes(&self) -> Result<(), Error> { + let buckets = format!("{}/{}", RUSTFS_META_BUCKET, BUCKET_META_PREFIX); + let multipart = format!("{}/{}", RUSTFS_META_BUCKET, "multipart"); + let config = format!("{}/{}", RUSTFS_META_BUCKET, "config"); + let tmp = format!("{}/{}", RUSTFS_META_BUCKET, "tmp"); + let defaults = vec![ + buckets.as_str(), + multipart.as_str(), + config.as_str(), + tmp.as_str(), + ]; + + self.make_volumes(defaults).await + } + pub fn resolve_abs_path(&self, path: impl AsRef) -> Result { Ok(path.as_ref().absolutize_virtually(&self.root)?.into_owned()) } @@ -151,18 +169,34 @@ pub async fn read_file_exists( path: impl AsRef, ) -> Result<(Vec, Option), Error> { let p = path.as_ref(); - let meta = match fs::metadata(&p).await { - Ok(meta) => Some(meta), - Err(e) => match e.kind() { - std::io::ErrorKind::NotFound => None, - _ => return Err(e.into()), - }, + let (data, meta) = match read_file_all(&p).await { + Ok((data, meta)) => (data, Some(meta)), + Err(e) => { + if is_err(&e, &DiskError::FileNotFound) { + (Vec::new(), None) + } else { + return Err(e); + } + } }; - let mut data = Vec::new(); - if meta.is_some() { - data = fs::read(&p).await?; - } + // let mut data = Vec::new(); + // if meta.is_some() { + // data = fs::read(&p).await?; + // } + + Ok((data, meta)) +} + +pub async fn read_file_all(path: impl AsRef) -> Result<(Vec, Metadata), Error> { + let p = path.as_ref(); + let meta = fs::metadata(&p).await.map_err(|e| match e.kind() { + ErrorKind::NotFound => Error::new(DiskError::FileNotFound), + ErrorKind::PermissionDenied => Error::new(DiskError::FileAccessDenied), + _ => Error::new(e), + })?; + + let data = fs::read(&p).await?; Ok((data, meta)) } @@ -172,17 +206,45 @@ impl DiskAPI for LocalDisk { #[must_use] async fn read_all(&self, volume: &str, path: &str) -> Result, Error> { let p = self.get_object_path(&volume, &path)?; - if p.exists() {} - let content = fs::read(p).await?; + let (data, _) = read_file_all(&p).await?; - unimplemented!() + Ok(data) + } + + async fn make_volumes(&self, volumes: Vec<&str>) -> Result<(), Error> { + for vol in volumes { + if let Err(e) = self.make_volume(vol).await { + match &e.downcast_ref::() { + Some(DiskError::VolumeExists) => Ok(()), + Some(_) => Err(e), + None => Err(e), + }?; + } + // TODO: health check + } + Ok(()) + } + async fn make_volume(&self, volume: &str) -> Result<(), Error> { + let p = self.get_bucket_path(&volume)?; + match File::open(&p).await { + Ok(_) => (), + Err(e) => match e.kind() { + ErrorKind::NotFound => { + fs::create_dir_all(&p).await?; + return Ok(()); + } + _ => return Err(Error::new(e)), + }, + } + + Err(Error::new(DiskError::VolumeExists)) } } pub struct RemoteDisk {} impl RemoteDisk { - pub fn new(ep: &Endpoint, health_check: bool) -> Result { + pub fn new(_ep: &Endpoint, _health_check: bool) -> Result { Ok(Self {}) } } @@ -190,12 +252,144 @@ impl RemoteDisk { #[async_trait::async_trait] pub trait DiskAPI { async fn read_all(&self, volume: &str, path: &str) -> Result, Error>; + + async fn make_volumes(&self, volume: Vec<&str>) -> Result<(), Error>; + async fn make_volume(&self, volume: &str) -> Result<(), Error>; } #[derive(Debug, thiserror::Error)] pub enum DiskError { + #[error("file not found")] + FileNotFound, #[error("disk not found")] DiskNotFound, + + #[error("disk access denied")] + FileAccessDenied, + #[error("InconsistentDisk")] InconsistentDisk, + + #[error("volume already exists")] + VolumeExists, + + #[error("unformatted disk error")] + UnformattedDisk, + + #[error("unsupport disk")] + UnsupportedDisk, + + #[error("disk not a dir")] + DiskNotDir, +} + +impl PartialEq for DiskError { + fn eq(&self, other: &Self) -> bool { + core::mem::discriminant(self) == core::mem::discriminant(other) + } +} + +pub fn check_disk_fatal_errs(errs: &Vec>) -> Result<(), Error> { + if count_errs(errs, &DiskError::UnsupportedDisk) == errs.len() { + return Err(Error::new(DiskError::UnsupportedDisk)); + } + + // if count_errs(errs, &DiskError::DiskAccessDenied) == errs.len() { + // return Err(Error::new(DiskError::DiskAccessDenied)); + // } + + if count_errs(errs, &DiskError::FileAccessDenied) == errs.len() { + return Err(Error::new(DiskError::FileAccessDenied)); + } + + // if count_errs(errs, &DiskError::FaultyDisk) == errs.len() { + // return Err(Error::new(DiskError::FaultyDisk)); + // } + + if count_errs(errs, &DiskError::DiskNotDir) == errs.len() { + return Err(Error::new(DiskError::DiskNotDir)); + } + + // if count_errs(errs, &DiskError::XLBackend) == errs.len() { + // return Err(Error::new(DiskError::XLBackend)); + // } + Ok(()) +} + +pub fn count_errs(errs: &Vec>, err: &DiskError) -> usize { + return errs + .iter() + .filter(|&e| { + if e.is_some() { + let e = e.as_ref().unwrap(); + let cast = e.downcast_ref::(); + if cast.is_some() { + let cast = cast.unwrap(); + return cast == err; + } + } + true + }) + .count(); +} + +pub fn is_err(err: &Error, disk_err: &DiskError) -> bool { + let cast = err.downcast_ref::(); + if cast.is_none() { + return false; + } + + let e = cast.unwrap(); + + e == disk_err +} + +pub fn match_err(err: Error, matchs: Vec) -> bool { + let cast = err.downcast_ref::(); + if cast.is_none() { + return false; + } + + let e = cast.unwrap(); + + for i in matchs.iter() { + if e == i { + return true; + } + } + + return false; +} + +#[cfg(test)] +mod test { + + use super::*; + + #[tokio::test] + async fn test_check_errs() {} + + #[tokio::test] + async fn test_make_volume() { + let p = "./testv"; + fs::create_dir_all(&p).await.unwrap(); + + let ep = match Endpoint::new(&p) { + Ok(e) => e, + Err(e) => { + println!("{e}"); + return; + } + }; + + let disk = LocalDisk::new(&ep, false).await.unwrap(); + + let volumes = vec!["a", "b", "c"]; + + disk.make_volumes(volumes.clone()).await.unwrap(); + + disk.make_volumes(volumes.clone()).await.unwrap(); + + fs::remove_dir_all(&p).await.unwrap(); + } } diff --git a/ecstore/src/endpoint.rs b/ecstore/src/endpoint.rs index 00136663b..49b32e35c 100644 --- a/ecstore/src/endpoint.rs +++ b/ecstore/src/endpoint.rs @@ -91,7 +91,7 @@ fn is_host_ip(ip_str: &str) -> bool { } impl Endpoint { - fn new(arg: &str) -> Result { + pub fn new(arg: &str) -> Result { if is_empty_path(arg) { return Err(Error::msg("不支持空或根endpoint")); } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index ad5c0fea3..357ed4ec5 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -1,7 +1,8 @@ +use futures::future::join_all; use uuid::Uuid; use crate::{ - disk::{self, DiskAPI, DiskOption}, + disk::{self, DiskAPI, DiskError, DiskOption, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET}, disks_layout::DisksLayout, endpoint::create_server_endpoints, format::FormatV3, @@ -43,13 +44,64 @@ impl ECStore { }) } - async fn load_formats( - disks: Vec>, - heal: bool, - ) -> (Vec, Vec) { + async fn init_format(disks: Vec>) -> Result { + let (formats, errs) = Self::load_format_all(&disks, false).await; unimplemented!() } + async fn load_format_all( + disks: &Vec>, + heal: bool, + ) -> (Vec>, Vec>) { + let mut futures = Vec::with_capacity(disks.len()); + + for ep in disks.iter() { + futures.push(Self::load_format(ep, heal)); + } + + let mut datas = Vec::with_capacity(disks.len()); + let mut errors = Vec::with_capacity(disks.len()); + + let results = join_all(futures).await; + for result in results { + match result { + Ok(s) => { + datas.push(Some(s)); + errors.push(None); + } + Err(e) => { + datas.push(None); + errors.push(Some(e)); + } + } + } + + (datas, errors) + } + + async fn load_format(disk: &Option, heal: bool) -> Result { + if disk.is_none() { + return Err(Error::new(DiskError::DiskNotFound)); + } + let disk = disk.as_ref().unwrap(); + + let data = disk + .read_all(RUSTFS_META_BUCKET, FORMAT_CONFIG_FILE) + .await + .map_err(|e| match &e.downcast_ref::() { + Some(DiskError::FileNotFound) => Error::new(DiskError::UnformattedDisk), + Some(DiskError::DiskNotFound) => Error::new(DiskError::UnformattedDisk), + Some(_) => e, + None => e, + })?; + + let fm = FormatV3::try_from(data.as_slice())?; + + // TODO: heal + + Ok(fm) + } + fn default_partiy_blocks(drive: usize) -> usize { match drive { 1 => 0,