init:store

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<impl DiskAPI, Error> {
@@ -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<Path>) -> Result<PathBuf, Error> {
         Ok(path.as_ref().absolutize_virtually(&self.root)?.into_owned())
     }
@@ -151,18 +169,34 @@ pub async fn read_file_exists(
     path: impl AsRef<Path>,
 ) -> Result<(Vec<u8>, Option<Metadata>), Error> {
     let p = path.as_ref();
-    let meta = match fs::metadata(&p).await {
+    let (data, meta) = match read_file_all(&p).await {
-        Ok(meta) => Some(meta),
+        Ok((data, meta)) => (data, Some(meta)),
-        Err(e) => match e.kind() {
+        Err(e) => {
-            std::io::ErrorKind::NotFound => None,
+            if is_err(&e, &DiskError::FileNotFound) {
-            _ => return Err(e.into()),
+                (Vec::new(), None)
-        },
+            } else {
+                return Err(e);
+            }
+        }
     };
 
-    let mut data = Vec::new();
+    // let mut data = Vec::new();
-    if meta.is_some() {
+    // if meta.is_some() {
-        data = fs::read(&p).await?;
+    //     data = fs::read(&p).await?;
-    }
+    // }
+
+    Ok((data, meta))
+}
+
+pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, 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<Vec<u8>, Error> {
         let p = self.get_object_path(&volume, &path)?;
-        if p.exists() {}
+        let (data, _) = read_file_all(&p).await?;
-        let content = fs::read(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::<DiskError>() {
+                    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<Self, Error> {
+    pub fn new(_ep: &Endpoint, _health_check: bool) -> Result<Self, Error> {
         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<Vec<u8>, 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<Option<Error>>) -> 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<Option<Error>>, err: &DiskError) -> usize {
+    return errs
+        .iter()
+        .filter(|&e| {
+            if e.is_some() {
+                let e = e.as_ref().unwrap();
+                let cast = e.downcast_ref::<DiskError>();
+                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::<DiskError>();
+    if cast.is_none() {
+        return false;
+    }
+
+    let e = cast.unwrap();
+
+    e == disk_err
+}
+
+pub fn match_err(err: Error, matchs: Vec<DiskError>) -> bool {
+    let cast = err.downcast_ref::<DiskError>();
+    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();
+    }
 }

ecstore/src/endpoint.rs

@@ -91,7 +91,7 @@ fn is_host_ip(ip_str: &str) -> bool {
 }
 
 impl Endpoint {
-    fn new(arg: &str) -> Result<Self, Error> {
+    pub fn new(arg: &str) -> Result<Self, Error> {
         if is_empty_path(arg) {
             return Err(Error::msg("不支持空或根endpoint"));
         }

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(
+    async fn init_format(disks: Vec<Option<impl DiskAPI>>) -> Result<FormatV3, Error> {
-        disks: Vec<Option<impl DiskAPI>>,
+        let (formats, errs) = Self::load_format_all(&disks, false).await;
-        heal: bool,
-    ) -> (Vec<FormatV3>, Vec<Error>) {
         unimplemented!()
     }
 
+    async fn load_format_all(
+        disks: &Vec<Option<impl DiskAPI>>,
+        heal: bool,
+    ) -> (Vec<Option<FormatV3>>, Vec<Option<Error>>) {
+        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<impl DiskAPI>, heal: bool) -> Result<FormatV3, Error> {
+        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::<DiskError>() {
+                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,