use anyhow:Result

This commit is contained in:
weisd
2024-06-28 13:37:56 +08:00
parent a9e515e5c0
commit 4c7c3d4c96
10 changed files with 64 additions and 58 deletions
+20 -22
View File
@@ -3,7 +3,7 @@ use std::{
path::{Path, PathBuf},
};
use anyhow::Error;
use anyhow::{Error, Result};
use bytes::Bytes;
use futures::future::join_all;
use path_absolutize::Absolutize;
@@ -32,7 +32,7 @@ pub struct DiskOption {
pub health_check: bool,
}
pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore, Error> {
pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
if ep.is_local {
Ok(LocalDisk::new(ep, opt.cleanup).await?)
} else {
@@ -71,7 +71,7 @@ pub async fn init_disks(
(storages, errors)
}
// pub async fn load_format(&self, heal: bool) -> Result<FormatV3, Error> {
// pub async fn load_format(&self, heal: bool) -> Result<FormatV3> {
// unimplemented!()
// }
@@ -87,7 +87,7 @@ pub struct LocalDisk {
}
impl LocalDisk {
pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Box<Self>, Error> {
pub async fn new(ep: &Endpoint, cleanup: bool) -> Result<Box<Self>> {
let root = fs::canonicalize(ep.url.path()).await?;
if cleanup {
@@ -134,7 +134,7 @@ impl LocalDisk {
Ok(Box::new(disk))
}
async fn make_meta_volumes(&self) -> Result<(), Error> {
async fn make_meta_volumes(&self) -> Result<()> {
let buckets = format!("{}/{}", RUSTFS_META_BUCKET, BUCKET_META_PREFIX);
let multipart = format!("{}/{}", RUSTFS_META_BUCKET, "multipart");
let config = format!("{}/{}", RUSTFS_META_BUCKET, "config");
@@ -149,22 +149,22 @@ impl LocalDisk {
self.make_volumes(defaults).await
}
pub fn resolve_abs_path(&self, path: impl AsRef<Path>) -> Result<PathBuf, Error> {
pub fn resolve_abs_path(&self, path: impl AsRef<Path>) -> Result<PathBuf> {
Ok(path.as_ref().absolutize_virtually(&self.root)?.into_owned())
}
pub fn get_object_path(&self, bucket: &str, key: &str) -> Result<PathBuf, Error> {
pub fn get_object_path(&self, bucket: &str, key: &str) -> Result<PathBuf> {
let dir = Path::new(&bucket);
let file_path = Path::new(&key);
self.resolve_abs_path(dir.join(file_path))
}
pub fn get_bucket_path(&self, bucket: &str) -> Result<PathBuf, Error> {
pub fn get_bucket_path(&self, bucket: &str) -> Result<PathBuf> {
let dir = Path::new(&bucket);
self.resolve_abs_path(dir)
}
// pub async fn load_format(&self) -> Result<Option<FormatV3>, Error> {
// pub async fn load_format(&self) -> Result<Option<FormatV3>> {
// let p = self.get_object_path(RUSTFS_META_BUCKET, FORMAT_CONFIG_FILE)?;
// let content = fs::read(&p).await?;
@@ -173,9 +173,7 @@ impl LocalDisk {
}
// 过滤 std::io::ErrorKind::NotFound
pub async fn read_file_exists(
path: impl AsRef<Path>,
) -> Result<(Vec<u8>, Option<Metadata>), Error> {
pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<Metadata>)> {
let p = path.as_ref();
let (data, meta) = match read_file_all(&p).await {
Ok((data, meta)) => (data, Some(meta)),
@@ -196,7 +194,7 @@ pub async fn read_file_exists(
Ok((data, meta))
}
pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, Metadata), Error> {
pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, Metadata)> {
let p = path.as_ref();
let meta = read_file_metadata(&path).await?;
@@ -205,7 +203,7 @@ pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, Metadata)
Ok((data, meta))
}
pub async fn read_file_metadata(p: impl AsRef<Path>) -> Result<Metadata, Error> {
pub async fn read_file_metadata(p: impl AsRef<Path>) -> Result<Metadata> {
let meta = fs::metadata(&p).await.map_err(|e| match e.kind() {
ErrorKind::NotFound => Error::new(DiskError::FileNotFound),
ErrorKind::PermissionDenied => Error::new(DiskError::FileAccessDenied),
@@ -215,7 +213,7 @@ pub async fn read_file_metadata(p: impl AsRef<Path>) -> Result<Metadata, Error>
Ok(meta)
}
pub async fn check_volume_exists(p: impl AsRef<Path>) -> Result<(), Error> {
pub async fn check_volume_exists(p: impl AsRef<Path>) -> Result<()> {
fs::metadata(&p).await.map_err(|e| match e.kind() {
ErrorKind::NotFound => Error::new(DiskError::VolumeNotFound),
ErrorKind::PermissionDenied => Error::new(DiskError::FileAccessDenied),
@@ -244,14 +242,14 @@ fn skip_access_checks(p: impl AsRef<str>) -> bool {
#[async_trait::async_trait]
impl DiskAPI for LocalDisk {
#[must_use]
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes, Error> {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
let p = self.get_object_path(&volume, &path)?;
let (data, _) = read_file_all(&p).await?;
Ok(Bytes::from(data))
}
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<(), Error> {
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
let p = self.get_object_path(&volume, &path)?;
// create top dir if not exists
@@ -267,7 +265,7 @@ impl DiskAPI for LocalDisk {
src_path: &str,
dst_volume: &str,
dst_path: &str,
) -> Result<(), Error> {
) -> Result<()> {
if !skip_access_checks(&src_volume) {
check_volume_exists(&src_volume).await?;
}
@@ -306,7 +304,7 @@ impl DiskAPI for LocalDisk {
Ok(())
}
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<(), Error> {
async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> {
for vol in volumes {
if let Err(e) = self.make_volume(vol).await {
match &e.downcast_ref::<DiskError>() {
@@ -319,7 +317,7 @@ impl DiskAPI for LocalDisk {
}
Ok(())
}
async fn make_volume(&self, volume: &str) -> Result<(), Error> {
async fn make_volume(&self, volume: &str) -> Result<()> {
let p = self.get_bucket_path(&volume)?;
match File::open(&p).await {
Ok(_) => (),
@@ -339,7 +337,7 @@ impl DiskAPI for LocalDisk {
// 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> {
// Ok(Self {})
// }
// }
@@ -374,7 +372,7 @@ pub enum DiskError {
}
impl DiskError {
pub fn check_disk_fatal_errs(errs: &Vec<Option<Error>>) -> Result<(), Error> {
pub fn check_disk_fatal_errs(errs: &Vec<Option<Error>>) -> Result<()> {
println!("errs: {:?}", errs);
if Self::count_errs(errs, &DiskError::UnsupportedDisk) == errs.len() {
+6 -6
View File
@@ -1,20 +1,20 @@
use std::fmt::Debug;
use anyhow::Error;
use anyhow::Result;
use bytes::Bytes;
#[async_trait::async_trait]
pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes, Error>;
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<(), Error>;
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>;
async fn rename_file(
&self,
src_volume: &str,
src_path: &str,
dst_volume: &str,
dst_path: &str,
) -> Result<(), Error>;
) -> Result<()>;
async fn make_volumes(&self, volume: Vec<&str>) -> Result<(), Error>;
async fn make_volume(&self, volume: &str) -> Result<(), Error>;
async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>;
async fn make_volume(&self, volume: &str) -> Result<()>;
}
+6 -6
View File
@@ -1,6 +1,6 @@
use std::collections::{HashMap, HashSet};
use anyhow::Error;
use anyhow::{Error, Result};
use serde::Deserialize;
use super::ellipses::*;
@@ -18,7 +18,7 @@ pub struct DisksLayout {
}
impl DisksLayout {
pub fn new(args: &Vec<String>) -> Result<DisksLayout, Error> {
pub fn new(args: &Vec<String>) -> Result<DisksLayout> {
if args.is_empty() {
return Err(Error::msg("Invalid argument"));
}
@@ -67,7 +67,7 @@ impl DisksLayout {
}
}
fn get_all_sets(set_drive_count: usize, args: &Vec<String>) -> Result<Vec<Vec<String>>, Error> {
fn get_all_sets(set_drive_count: usize, args: &Vec<String>) -> Result<Vec<Vec<String>>> {
let set_args;
if !has_ellipses(args) {
let set_indexes: Vec<Vec<usize>>;
@@ -114,7 +114,7 @@ pub struct EndpointSet {
}
impl EndpointSet {
pub fn new(args: &Vec<String>, set_div_count: usize) -> Result<EndpointSet, Error> {
pub fn new(args: &Vec<String>, set_div_count: usize) -> Result<EndpointSet> {
let mut arg_patterns = Vec::with_capacity(args.len());
for arg in args.iter() {
arg_patterns.push(find_ellipses_patterns(arg.as_str())?);
@@ -164,7 +164,7 @@ impl EndpointSet {
}
}
// fn parse_endpoint_set(set_div_count: usize, args: &Vec<String>) -> Result<EndpointSet, Error> {
// fn parse_endpoint_set(set_div_count: usize, args: &Vec<String>) -> Result<EndpointSet> {
// let mut arg_patterns = Vec::with_capacity(args.len());
// for arg in args.iter() {
// arg_patterns.push(find_ellipses_patterns(arg.as_str())?);
@@ -267,7 +267,7 @@ fn get_set_indexes(
totalsizes: &Vec<usize>,
set_div_count: usize,
arg_patterns: &Vec<ArgPattern>,
) -> Result<Vec<Vec<usize>>, Error> {
) -> Result<Vec<Vec<usize>>> {
if args.is_empty() || totalsizes.is_empty() {
return Err(Error::msg("Invalid argument"));
}
+3 -3
View File
@@ -1,6 +1,6 @@
use lazy_static::*;
use anyhow::Error;
use anyhow::{Error, Result};
use regex::Regex;
lazy_static! {
@@ -89,7 +89,7 @@ impl ArgPattern {
}
#[allow(dead_code)]
pub fn find_ellipses_patterns(arg: &str) -> Result<ArgPattern, Error> {
pub fn find_ellipses_patterns(arg: &str) -> Result<ArgPattern> {
let mut caps = match ELLIPSES_RE.captures(arg) {
Some(caps) => caps,
None => return Err(Error::msg("Invalid argument")),
@@ -175,7 +175,7 @@ pub fn has_ellipses(s: &Vec<String>) -> bool {
// `{1...64}`
// `{33...64}`
#[allow(dead_code)]
pub fn parse_ellipses_range(partten: &str) -> Result<Vec<String>, Error> {
pub fn parse_ellipses_range(partten: &str) -> Result<Vec<String>> {
if !partten.contains(OPEN_BRACES) {
return Err(Error::msg("Invalid argument"));
}
+9 -9
View File
@@ -3,7 +3,7 @@ use super::utils::{
net::{is_local_host, split_host_port},
string::new_string_set,
};
use anyhow::Error;
use anyhow::{Error, Result};
use std::fmt::Display;
use std::{collections::HashMap, net::IpAddr, path::Path, usize};
use url::{ParseError, Url};
@@ -91,7 +91,7 @@ fn is_host_ip(ip_str: &str) -> bool {
}
impl Endpoint {
pub fn new(arg: &str) -> Result<Self, Error> {
pub fn new(arg: &str) -> Result<Self> {
if is_empty_path(arg) {
return Err(Error::msg("不支持空或根endpoint"));
}
@@ -207,7 +207,7 @@ impl Endpoint {
self.disk_idx = idx
}
fn update_islocal(&mut self) -> Result<(), Error> {
fn update_islocal(&mut self) -> Result<()> {
if self.url.has_host() {
self.is_local = is_local_host(
self.url.host().unwrap(),
@@ -253,7 +253,7 @@ impl Endpoints {
pub fn slice(&self, start: usize, end: usize) -> Vec<Endpoint> {
self.0.as_slice()[start..end].to_vec()
}
pub fn from_args(args: Vec<String>) -> Result<Self, Error> {
pub fn from_args(args: Vec<String>) -> Result<Self> {
let mut ep_type = EndpointType::UnKnow;
let mut scheme = String::new();
let mut eps = Vec::new();
@@ -297,7 +297,7 @@ impl PoolEndpointList {
}
// TODO: 解析域名,判断哪个是本地地址
fn update_is_local(&mut self) -> Result<(), Error> {
fn update_is_local(&mut self) -> Result<()> {
for eps in self.0.iter_mut() {
for ep in eps.iter_mut() {
// TODO:
@@ -336,7 +336,7 @@ impl EndpointServerPools {
server_addr: String,
pool_args: &Vec<PoolDisksLayout>,
legacy: bool,
) -> Result<(EndpointServerPools, SetupType), Error> {
) -> Result<(EndpointServerPools, SetupType)> {
if pool_args.is_empty() {
return Err(Error::msg("无效参数"));
}
@@ -380,7 +380,7 @@ impl EndpointServerPools {
self.0.push(pes)
}
pub fn add(&mut self, eps: PoolEndpoints) -> Result<(), Error> {
pub fn add(&mut self, eps: PoolEndpoints) -> Result<()> {
let mut exits = new_string_set();
for peps in self.0.iter() {
for ep in peps.endpoints.0.iter() {
@@ -472,7 +472,7 @@ fn is_single_drive_layout(pools_layout: &Vec<PoolDisksLayout>) -> bool {
pub fn create_pool_endpoints(
server_addr: String,
pools: &Vec<PoolDisksLayout>,
) -> Result<(Vec<Endpoints>, SetupType), Error> {
) -> Result<(Vec<Endpoints>, SetupType)> {
if is_empty_layout(pools) {
return Err(Error::msg("empty layout"));
}
@@ -571,7 +571,7 @@ fn create_server_endpoints(
server_addr: String,
pool_args: &Vec<PoolDisksLayout>,
legacy: bool,
) -> Result<(EndpointServerPools, SetupType), Error> {
) -> Result<(EndpointServerPools, SetupType)> {
if pool_args.is_empty() {
return Err(Error::msg("无效参数"));
}
+2 -2
View File
@@ -1,4 +1,4 @@
use anyhow::Error;
use anyhow::{Error, Result};
use serde::{Deserialize, Serialize};
use serde_json::Error as JsonError;
use uuid::Uuid;
@@ -165,7 +165,7 @@ impl FormatV3 {
/// format, after successful validation.
/// - i'th position is the set index
/// - j'th position is the disk index in the current set
pub fn find_disk_index_by_disk_id(&self, disk_id: Uuid) -> Result<(usize, usize), Error> {
pub fn find_disk_index_by_disk_id(&self, disk_id: Uuid) -> Result<(usize, usize)> {
if disk_id == Uuid::nil() {
return Err(Error::new(disk::DiskError::DiskNotFound));
}
+2 -2
View File
@@ -1,4 +1,4 @@
use anyhow::Error;
use anyhow::Result;
use uuid::Uuid;
use crate::{endpoint::PoolEndpoints, format::FormatV3};
@@ -22,7 +22,7 @@ impl Sets {
fm: &FormatV3,
pool_idx: usize,
partiy_count: usize,
) -> Result<Self, Error> {
) -> Result<Self> {
let set_count = fm.erasure.sets.len();
let set_drive_count = fm.erasure.sets[0].len();
+8 -4
View File
@@ -1,6 +1,6 @@
use std::collections::HashMap;
use anyhow::Error;
use anyhow::{Error, Result};
use uuid::Uuid;
use crate::{
@@ -8,7 +8,7 @@ use crate::{
disks_layout::DisksLayout,
endpoint::EndpointServerPools,
sets::Sets,
store_api::StorageAPI,
store_api::{MakeBucketOptions, StorageAPI},
store_init,
};
@@ -22,7 +22,7 @@ pub struct ECStore {
}
impl ECStore {
pub async fn new(address: String, endpoints: Vec<String>) -> Result<Self, Error> {
pub async fn new(address: String, endpoints: Vec<String>) -> Result<Self> {
let layouts = DisksLayout::new(&endpoints)?;
let mut deployment_id = None;
@@ -88,7 +88,11 @@ impl ECStore {
}
impl StorageAPI for ECStore {
async fn put_object(&self, bucket: &str, objcet: &str) -> Result<(), Error> {
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> {
// TODO: check valid bucket name
unimplemented!()
}
async fn put_object(&self, bucket: &str, objcet: &str) -> Result<()> {
unimplemented!()
}
}
+6 -2
View File
@@ -1,5 +1,9 @@
use anyhow::Error;
use anyhow::Result;
pub struct MakeBucketOptions {}
pub trait StorageAPI {
async fn put_object(&self, bucket: &str, objcet: &str) -> Result<(), Error>;
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
async fn put_object(&self, bucket: &str, objcet: &str) -> Result<()>;
}
+2 -2
View File
@@ -3,11 +3,11 @@ use std::{
net::{IpAddr, ToSocketAddrs},
};
use anyhow::Error;
use anyhow::{Error, Result};
use netif;
use url::Host;
pub fn split_host_port(s: &str) -> Result<(String, u16), Error> {
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>() {