This commit is contained in:
weisd
2025-06-06 11:35:27 +08:00
parent 9384b831ec
commit db355bb26b
37 changed files with 2169 additions and 668 deletions
Generated
+3 -3
View File
@@ -1,6 +1,6 @@
# This file is automatically @generated by Cargo.
# It is not intended for manual editing.
version = 3
version = 4
[[package]]
name = "addr2line"
@@ -4393,7 +4393,6 @@ dependencies = [
"arc-swap",
"async-trait",
"base64-simd",
"common",
"crypto",
"ecstore",
"futures",
@@ -6748,7 +6747,6 @@ dependencies = [
"arc-swap",
"async-trait",
"base64-simd",
"common",
"crypto",
"futures",
"ipnetwork",
@@ -7710,6 +7708,7 @@ dependencies = [
"rust-embed",
"rustfs-config",
"rustfs-event-notifier",
"rustfs-filemeta",
"rustfs-obs",
"rustfs-utils",
"rustfs-zip",
@@ -7720,6 +7719,7 @@ dependencies = [
"serde_urlencoded",
"shadow-rs",
"socket2",
"thiserror 2.0.12",
"tikv-jemallocator",
"time",
"tokio",
+2 -2
View File
@@ -26,10 +26,10 @@ members = [
resolver = "2"
[workspace.package]
edition = "2021"
edition = "2024"
license = "Apache-2.0"
repository = "https://github.com/rustfs/rustfs"
rust-version = "1.75"
rust-version = "1.85"
version = "0.0.1"
[workspace.lints.rust]
+7
View File
@@ -169,3 +169,10 @@ impl From<rmp::decode::MarkerReadError> for Error {
Error::RmpDecodeMarkerRead(serr)
}
}
pub fn is_io_eof(e: &Error) -> bool {
match e {
Error::Io(e) => e.kind() == std::io::ErrorKind::UnexpectedEof,
_ => false,
}
}
+15 -25
View File
@@ -515,11 +515,7 @@ impl FileMeta {
let has_vid = {
if !version_id.is_empty() {
let id = Uuid::parse_str(version_id)?;
if !id.is_nil() {
Some(id)
} else {
None
}
if !id.is_nil() { Some(id) } else { None }
} else {
None
}
@@ -571,6 +567,16 @@ impl FileMeta {
versions.push(fi);
}
let mut prev_mod_time = None;
for (i, fi) in versions.iter_mut().enumerate() {
if i == 0 {
fi.is_latest = true;
} else {
fi.successor_mod_time = prev_mod_time;
}
prev_mod_time = fi.mod_time;
}
Ok(FileInfoVersions {
volume: volume.to_string(),
name: path.to_string(),
@@ -1225,11 +1231,7 @@ impl FileMetaVersionHeader {
cur.read_exact(&mut buf)?;
self.version_id = {
let id = Uuid::from_bytes(buf);
if id.is_nil() {
None
} else {
Some(id)
}
if id.is_nil() { None } else { Some(id) }
};
// mod_time
@@ -1403,11 +1405,7 @@ impl MetaObject {
cur.read_exact(&mut buf)?;
self.version_id = {
let id = Uuid::from_bytes(buf);
if id.is_nil() {
None
} else {
Some(id)
}
if id.is_nil() { None } else { Some(id) }
};
}
"DDir" => {
@@ -1416,11 +1414,7 @@ impl MetaObject {
cur.read_exact(&mut buf)?;
self.data_dir = {
let id = Uuid::from_bytes(buf);
if id.is_nil() {
None
} else {
Some(id)
}
if id.is_nil() { None } else { Some(id) }
};
}
"EcAlgo" => {
@@ -1986,11 +1980,7 @@ impl MetaDeleteMarker {
cur.read_exact(&mut buf)?;
self.version_id = {
let id = Uuid::from_bytes(buf);
if id.is_nil() {
None
} else {
Some(id)
}
if id.is_nil() { None } else { Some(id) }
};
}
+1 -1
View File
@@ -176,7 +176,7 @@ impl Stream for ReceiverStream {
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
let poll = Pin::new(&mut self.receiver).poll_recv(cx);
match &poll {
Poll::Ready(Some(Some(ref bytes))) => {
Poll::Ready(Some(Some(bytes))) => {
http_log!("[ReceiverStream] poll_next: got {} bytes", bytes.len());
}
Poll::Ready(Some(None)) => {
+1 -1
View File
@@ -443,7 +443,7 @@ impl StorageError {
pub fn from_u32(error: u32) -> Option<Self> {
match error {
0x01 => Some(StorageError::Io(std::io::Error::new(std::io::ErrorKind::Other, "Io error"))),
0x01 => Some(StorageError::Io(std::io::Error::other("Io error"))),
0x02 => Some(StorageError::FaultyDisk),
0x03 => Some(StorageError::DiskFull),
0x04 => Some(StorageError::VolumeNotFound),
+16 -31
View File
@@ -2,22 +2,22 @@
use crate::bucket::metadata_sys::{self, set_bucket_metadata};
use crate::bucket::utils::{check_valid_bucket_name, check_valid_bucket_name_strict, is_meta_bucketname};
use crate::config::storageclass;
use crate::config::GLOBAL_StorageClass;
use crate::config::storageclass;
use crate::disk::endpoint::{Endpoint, EndpointType};
use crate::disk::{DiskAPI, DiskInfo, DiskInfoOptions};
use crate::error::{
is_err_bucket_exists, is_err_invalid_upload_id, is_err_object_not_found, is_err_read_quorum, is_err_version_not_found,
to_object_err, StorageError,
StorageError, is_err_bucket_exists, is_err_invalid_upload_id, is_err_object_not_found, is_err_read_quorum,
is_err_version_not_found, to_object_err,
};
use crate::global::{
get_global_endpoints, is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, DISK_ASSUME_UNKNOWN_SIZE,
DISK_FILL_FRACTION, DISK_MIN_INODES, DISK_RESERVE_FRACTION, GLOBAL_BOOT_TIME, GLOBAL_LOCAL_DISK_MAP,
GLOBAL_LOCAL_DISK_SET_DRIVES,
DISK_ASSUME_UNKNOWN_SIZE, DISK_FILL_FRACTION, DISK_MIN_INODES, DISK_RESERVE_FRACTION, GLOBAL_BOOT_TIME,
GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, get_global_endpoints, is_dist_erasure, is_erasure_sd,
set_global_deployment_id, set_object_layer,
};
use crate::heal::data_usage::{DataUsageInfo, DATA_USAGE_ROOT};
use crate::heal::data_usage::{DATA_USAGE_ROOT, DataUsageInfo};
use crate::heal::data_usage_cache::{DataUsageCache, DataUsageCacheInfo};
use crate::heal::heal_commands::{HealOpts, HealScanMode, HEAL_ITEM_METADATA};
use crate::heal::heal_commands::{HEAL_ITEM_METADATA, HealOpts, HealScanMode};
use crate::heal::heal_ops::{HealEntryFn, HealSequence};
use crate::new_object_layer_fn;
use crate::notification_sys::get_global_notification_sys;
@@ -26,11 +26,11 @@ use crate::rebalance::RebalanceMeta;
use crate::store_api::{ListMultipartsInfo, ListObjectVersionsInfo, MultipartInfo, ObjectIO};
use crate::store_init::{check_disk_fatal_errs, ec_drives_no_config};
use crate::utils::crypto::base64_decode;
use crate::utils::path::{decode_dir_object, encode_dir_object, path_join_buf, SLASH_SEPARATOR};
use crate::utils::path::{SLASH_SEPARATOR, decode_dir_object, encode_dir_object, path_join_buf};
use crate::utils::xml;
use crate::{
bucket::metadata::BucketMetadata,
disk::{new_disk, DiskOption, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
disk::{BUCKET_META_PREFIX, DiskOption, DiskStore, RUSTFS_META_BUCKET, new_disk},
endpoints::EndpointServerPools,
peer::S3PeerSys,
sets::Sets,
@@ -60,7 +60,7 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
use time::OffsetDateTime;
use tokio::select;
use tokio::sync::mpsc::Sender;
use tokio::sync::{broadcast, mpsc, RwLock};
use tokio::sync::{RwLock, broadcast, mpsc};
use tokio::time::{interval, sleep};
use tracing::{debug, info};
use tracing::{error, warn};
@@ -502,8 +502,7 @@ impl ECStore {
return None;
}
let mut rng = rand::thread_rng();
let random_u64: u64 = rng.gen();
let random_u64: u64 = rand::random();
let choose = random_u64 % total;
let mut at_total = 0;
@@ -1965,11 +1964,7 @@ impl StorageAPI for ECStore {
Ok(_) => return Ok(()),
Err(err) => {
//
if is_err_invalid_upload_id(&err) {
None
} else {
Some(err)
}
if is_err_invalid_upload_id(&err) { None } else { Some(err) }
}
};
@@ -2010,11 +2005,7 @@ impl StorageAPI for ECStore {
Ok(res) => return Ok(res),
Err(err) => {
//
if is_err_invalid_upload_id(&err) {
None
} else {
Some(err)
}
if is_err_invalid_upload_id(&err) { None } else { Some(err) }
}
};
@@ -2251,7 +2242,7 @@ impl StorageAPI for ECStore {
HealSequence::heal_meta_object(hs_clone.clone(), &bucket, &entry.name, "", scan_mode).await
} else {
HealSequence::heal_object(hs_clone.clone(), &bucket, &entry.name, "", scan_mode).await
}
};
}
};
@@ -2624,13 +2615,7 @@ impl ServerPoolsAvailableSpace {
}
pub async fn has_space_for(dis: &[Option<DiskInfo>], size: i64) -> Result<bool> {
let size = {
if size < 0 {
DISK_ASSUME_UNKNOWN_SIZE
} else {
size as u64 * 2
}
};
let size = { if size < 0 { DISK_ASSUME_UNKNOWN_SIZE } else { size as u64 * 2 } };
let mut available = 0;
let mut total = 0;
+1 -1
View File
@@ -31,7 +31,7 @@ tracing.workspace = true
madmin.workspace = true
lazy_static.workspace = true
regex = "1.11.1"
common.workspace = true
[dev-dependencies]
test-case.workspace = true
+95 -44
View File
@@ -1,14 +1,12 @@
use ecstore::disk::error::DiskError;
use policy::policy::Error as PolicyError;
pub type Result<T> = core::result::Result<T, Error>;
#[derive(thiserror::Error, Debug)]
pub enum Error {
#[error(transparent)]
PolicyError(#[from] PolicyError),
#[error("ecstore error: {0}")]
EcstoreError(common::error::Error),
#[error("{0}")]
StringError(String),
@@ -91,71 +89,124 @@ pub enum Error {
#[error("policy too large")]
PolicyTooLarge,
#[error("config not found")]
ConfigNotFound,
#[error("io error: {0}")]
Io(std::io::Error),
}
impl Error {
pub fn other<E>(error: E) -> Self
where
E: Into<Box<dyn std::error::Error + Send + Sync>>,
{
Error::Io(std::io::Error::other(error))
}
}
impl From<ecstore::error::StorageError> for Error {
fn from(e: ecstore::error::StorageError) -> Self {
match e {
ecstore::error::StorageError::ConfigNotFound => Error::ConfigNotFound,
_ => Error::other(e),
}
}
}
impl From<policy::error::Error> for Error {
fn from(e: policy::error::Error) -> Self {
match e {
policy::error::Error::PolicyTooLarge => Error::PolicyTooLarge,
policy::error::Error::InvalidArgument => Error::InvalidArgument,
policy::error::Error::InvalidServiceType(s) => Error::InvalidServiceType(s),
policy::error::Error::IAMActionNotAllowed => Error::IAMActionNotAllowed,
policy::error::Error::InvalidExpiration => Error::InvalidExpiration,
policy::error::Error::NoAccessKey => Error::NoAccessKey,
policy::error::Error::InvalidToken => Error::InvalidToken,
policy::error::Error::InvalidAccessKey => Error::InvalidAccessKey,
policy::error::Error::NoSecretKeyWithAccessKey => Error::NoSecretKeyWithAccessKey,
policy::error::Error::NoAccessKeyWithSecretKey => Error::NoAccessKeyWithSecretKey,
policy::error::Error::Io(e) => Error::Io(e),
policy::error::Error::JWTError(e) => Error::JWTError(e),
policy::error::Error::NoSuchUser(s) => Error::NoSuchUser(s),
policy::error::Error::NoSuchAccount(s) => Error::NoSuchAccount(s),
policy::error::Error::NoSuchServiceAccount(s) => Error::NoSuchServiceAccount(s),
policy::error::Error::NoSuchTempAccount(s) => Error::NoSuchTempAccount(s),
policy::error::Error::NoSuchGroup(s) => Error::NoSuchGroup(s),
policy::error::Error::NoSuchPolicy => Error::NoSuchPolicy,
policy::error::Error::PolicyInUse => Error::PolicyInUse,
policy::error::Error::GroupNotEmpty => Error::GroupNotEmpty,
policy::error::Error::InvalidAccessKeyLength => Error::InvalidAccessKeyLength,
policy::error::Error::InvalidSecretKeyLength => Error::InvalidSecretKeyLength,
policy::error::Error::ContainsReservedChars => Error::ContainsReservedChars,
policy::error::Error::GroupNameContainsReservedChars => Error::GroupNameContainsReservedChars,
policy::error::Error::CredNotInitialized => Error::CredNotInitialized,
policy::error::Error::IamSysNotInitialized => Error::IamSysNotInitialized,
policy::error::Error::PolicyError(e) => Error::PolicyError(e),
policy::error::Error::StringError(s) => Error::StringError(s),
policy::error::Error::CryptoError(e) => Error::CryptoError(e),
policy::error::Error::ErrCredMalformed => Error::ErrCredMalformed,
}
}
}
impl From<serde_json::Error> for Error {
fn from(e: serde_json::Error) -> Self {
Error::other(e)
}
}
impl From<base64_simd::Error> for Error {
fn from(e: base64_simd::Error) -> Self {
Error::other(e)
}
}
pub fn is_err_config_not_found(err: &Error) -> bool {
matches!(err, Error::ConfigNotFound)
}
// pub fn is_err_no_such_user(e: &Error) -> bool {
// matches!(e, Error::NoSuchUser(_))
// }
pub fn is_err_no_such_policy(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchPolicy)
} else {
false
}
pub fn is_err_no_such_policy(err: &Error) -> bool {
matches!(err, Error::NoSuchPolicy)
}
pub fn is_err_no_such_user(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchUser(_))
} else {
false
}
pub fn is_err_no_such_user(err: &Error) -> bool {
matches!(err, Error::NoSuchUser(_))
}
pub fn is_err_no_such_account(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchAccount(_))
} else {
false
}
pub fn is_err_no_such_account(err: &Error) -> bool {
matches!(err, Error::NoSuchAccount(_))
}
pub fn is_err_no_such_temp_account(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchTempAccount(_))
} else {
false
}
pub fn is_err_no_such_temp_account(err: &Error) -> bool {
matches!(err, Error::NoSuchTempAccount(_))
}
pub fn is_err_no_such_group(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchGroup(_))
} else {
false
}
pub fn is_err_no_such_group(err: &Error) -> bool {
matches!(err, Error::NoSuchGroup(_))
}
pub fn is_err_no_such_service_account(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchServiceAccount(_))
} else {
false
}
pub fn is_err_no_such_service_account(err: &Error) -> bool {
matches!(err, Error::NoSuchServiceAccount(_))
}
// pub fn clone_err(e: &common::error::Error) -> common::error::Error {
// pub fn clone_err(e: &Error) -> Error {
// if let Some(e) = e.downcast_ref::<DiskError>() {
// clone_disk_err(e)
// } else if let Some(e) = e.downcast_ref::<std::io::Error>() {
// if let Some(code) = e.raw_os_error() {
// common::error::Error::new(std::io::Error::from_raw_os_error(code))
// Error::new(std::io::Error::from_raw_os_error(code))
// } else {
// common::error::Error::new(std::io::Error::new(e.kind(), e.to_string()))
// Error::new(std::io::Error::new(e.kind(), e.to_string()))
// }
// } else {
// //TODO: Optimize other types
// common::error::Error::msg(e.to_string())
// Error::msg(e.to_string())
// }
// }
+2 -6
View File
@@ -1,6 +1,5 @@
use common::error::{Error, Result};
use crate::error::{Error, Result};
use ecstore::store::ECStore;
use error::Error as IamError;
use manager::IamCache;
use policy::auth::Credentials;
use std::sync::{Arc, OnceLock};
@@ -62,8 +61,5 @@ pub async fn init_iam_sys(ecstore: Arc<ECStore>) -> Result<()> {
#[inline]
pub fn get() -> Result<Arc<IamSys<ObjectStore>>> {
IAM_SYS
.get()
.map(Arc::clone)
.ok_or(Error::new(IamError::IamSysNotInitialized))
IAM_SYS.get().map(Arc::clone).ok_or(Error::IamSysNotInitialized)
}
+47 -52
View File
@@ -1,3 +1,4 @@
use crate::error::{is_err_config_not_found, Error, Result};
use crate::{
cache::{Cache, CacheEntity},
error::{is_err_no_such_group, is_err_no_such_policy, is_err_no_such_user, Error as IamError},
@@ -8,7 +9,6 @@ use crate::{
STATUS_DISABLED, STATUS_ENABLED,
},
};
use common::error::{Error, Result};
use ecstore::utils::{crypto::base64_encode, path::path_join_buf};
use madmin::{AccountStatus, AddOrUpdateUserReq, GroupDesc};
use policy::{
@@ -182,7 +182,7 @@ where
pub async fn get_policy(&self, name: &str) -> Result<Policy> {
if name.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let policies = MappedPolicy::new(name).to_slice();
@@ -199,13 +199,13 @@ where
.load()
.get(&policy)
.cloned()
.ok_or(Error::new(IamError::NoSuchPolicy))?;
.ok_or(Error::NoSuchPolicy)?;
to_merge.push(v.policy);
}
if to_merge.is_empty() {
return Err(Error::new(IamError::NoSuchPolicy));
return Err(Error::NoSuchPolicy);
}
Ok(Policy::merge_policies(to_merge))
@@ -213,20 +213,15 @@ where
pub async fn get_policy_doc(&self, name: &str) -> Result<PolicyDoc> {
if name.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
self.cache
.policy_docs
.load()
.get(name)
.cloned()
.ok_or(Error::new(IamError::NoSuchPolicy))
self.cache.policy_docs.load().get(name).cloned().ok_or(Error::NoSuchPolicy)
}
pub async fn delete_policy(&self, name: &str, is_from_notify: bool) -> Result<()> {
if name.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
if is_from_notify {
@@ -254,7 +249,7 @@ where
});
if !users.is_empty() || !groups.is_empty() {
return Err(IamError::PolicyInUse.into());
return Err(Error::PolicyInUse);
}
if let Err(err) = self.api.delete_policy_doc(name).await {
@@ -274,7 +269,7 @@ where
pub async fn set_policy(&self, name: &str, policy: Policy) -> Result<OffsetDateTime> {
if name.is_empty() || policy.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let policy_doc = self
@@ -406,7 +401,7 @@ where
}
if !user_exists {
return Err(Error::new(IamError::NoSuchUser(access_key.to_string())));
return Err(Error::NoSuchUser(access_key.to_string()));
}
Ok(ret)
@@ -452,13 +447,13 @@ where
/// create a service account and update cache
pub async fn add_service_account(&self, cred: Credentials) -> Result<OffsetDateTime> {
if cred.access_key.is_empty() || cred.parent_user.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let users = self.cache.users.load();
if let Some(x) = users.get(&cred.access_key) {
if x.credentials.is_service_account() {
return Err(Error::new(IamError::IAMActionNotAllowed));
return Err(Error::IAMActionNotAllowed);
}
}
@@ -475,11 +470,11 @@ where
pub async fn update_service_account(&self, name: &str, opts: UpdateServiceAccountOpts) -> Result<OffsetDateTime> {
let Some(ui) = self.cache.users.load().get(name).cloned() else {
return Err(IamError::NoSuchServiceAccount(name.to_string()).into());
return Err(Error::NoSuchServiceAccount(name.to_string()));
};
if !ui.credentials.is_service_account() {
return Err(IamError::NoSuchServiceAccount(name.to_string()).into());
return Err(Error::NoSuchServiceAccount(name.to_string()));
}
let mut cr = ui.credentials.clone();
@@ -487,7 +482,7 @@ where
if let Some(secret) = opts.secret_key {
if !is_secret_key_valid(&secret) {
return Err(IamError::InvalidSecretKeyLength.into());
return Err(Error::InvalidSecretKeyLength);
}
cr.secret_key = secret;
}
@@ -534,7 +529,7 @@ where
if !session_policy.version.is_empty() && !session_policy.statements.is_empty() {
let policy_buf = serde_json::to_vec(&session_policy)?;
if policy_buf.len() > MAX_SVCSESSION_POLICY_SIZE {
return Err(IamError::PolicyTooLarge.into());
return Err(Error::PolicyTooLarge);
}
m.insert(SESSION_POLICY_NAME.to_owned(), serde_json::Value::String(base64_encode(&policy_buf)));
@@ -557,7 +552,7 @@ where
pub async fn policy_db_get(&self, name: &str, groups: &Option<Vec<String>>) -> Result<Vec<String>> {
if name.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let (mut policies, _) = self.policy_db_get_internal(name, false, false).await?;
@@ -593,7 +588,7 @@ where
Cache::add_or_update(&self.cache.groups, name, p, OffsetDateTime::now_utc());
}
m.get(name).cloned().ok_or(IamError::NoSuchGroup(name.to_string()))?
m.get(name).cloned().ok_or(Error::NoSuchGroup(name.to_string()))?
}
};
@@ -736,7 +731,7 @@ where
}
pub async fn policy_db_set(&self, name: &str, user_type: UserType, is_group: bool, policy: &str) -> Result<OffsetDateTime> {
if name.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
if policy.is_empty() {
@@ -762,7 +757,7 @@ where
let policy_docs_cache = self.cache.policy_docs.load();
for p in mp.to_slice() {
if !policy_docs_cache.contains_key(&p) {
return Err(Error::new(IamError::NoSuchPolicy));
return Err(Error::NoSuchPolicy);
}
}
@@ -790,14 +785,14 @@ where
cred.is_expired(),
cred.parent_user.is_empty()
);
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
if let Some(policy) = policy_name {
let mp = MappedPolicy::new(policy);
let (_, combined_policy_stmt) = filter_policies(&self.cache, &mp.policies, "temp");
if combined_policy_stmt.is_empty() {
return Err(Error::msg(format!("need poliy not found {}", IamError::NoSuchPolicy)));
return Err(Error::other(format!("need poliy not found {}", IamError::NoSuchPolicy)));
}
self.api
@@ -824,11 +819,11 @@ where
let u = match users.get(name) {
Some(u) => u,
None => return Err(Error::new(IamError::NoSuchUser(name.to_string()))),
None => return Err(Error::NoSuchUser(name.to_string())),
};
if u.credentials.is_temp() || u.credentials.is_service_account() {
return Err(Error::new(IamError::IAMActionNotAllowed));
return Err(Error::IAMActionNotAllowed);
}
let mut uinfo = madmin::UserInfo {
@@ -960,7 +955,7 @@ where
if let Some(x) = users.get(access_key) {
warn!("user already exists: {:?}", x);
if x.credentials.is_temp() {
return Err(IamError::IAMActionNotAllowed.into());
return Err(Error::IAMActionNotAllowed);
}
}
@@ -988,7 +983,7 @@ where
pub async fn delete_user(&self, access_key: &str, utype: UserType) -> Result<()> {
if access_key.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
if utype == UserType::Reg {
@@ -1040,13 +1035,13 @@ where
pub async fn update_user_secret_key(&self, access_key: &str, secret_key: &str) -> Result<()> {
if access_key.is_empty() || secret_key.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let users = self.cache.users.load();
let u = match users.get(access_key) {
Some(u) => u,
None => return Err(Error::new(IamError::NoSuchUser(access_key.to_string()))),
None => return Err(Error::NoSuchUser(access_key.to_string())),
};
let mut cred = u.credentials.clone();
@@ -1063,21 +1058,21 @@ where
pub async fn set_user_status(&self, access_key: &str, status: AccountStatus) -> Result<OffsetDateTime> {
if access_key.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
if !access_key.is_empty() && status != AccountStatus::Enabled && status != AccountStatus::Disabled {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let users = self.cache.users.load();
let u = match users.get(access_key) {
Some(u) => u,
None => return Err(Error::new(IamError::NoSuchUser(access_key.to_string()))),
None => return Err(Error::NoSuchUser(access_key.to_string())),
};
if u.credentials.is_temp() || u.credentials.is_service_account() {
return Err(Error::new(IamError::IAMActionNotAllowed));
return Err(Error::IAMActionNotAllowed);
}
let status = {
@@ -1122,7 +1117,7 @@ where
let users = self.cache.users.load();
let u = match users.get(access_key) {
Some(u) => u,
None => return Err(Error::new(IamError::NoSuchUser(access_key.to_string()))),
None => return Err(Error::NoSuchUser(access_key.to_string())),
};
if u.credentials.is_temp() {
@@ -1134,7 +1129,7 @@ where
pub async fn add_users_to_group(&self, group: &str, members: Vec<String>) -> Result<OffsetDateTime> {
if group.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let users_cache = self.cache.users.load();
@@ -1142,10 +1137,10 @@ where
for member in members.iter() {
if let Some(u) = users_cache.get(member) {
if u.credentials.is_temp() || u.credentials.is_service_account() {
return Err(Error::new(IamError::IAMActionNotAllowed));
return Err(Error::IAMActionNotAllowed);
}
} else {
return Err(Error::new(IamError::NoSuchUser(member.to_string())));
return Err(Error::NoSuchUser(member.to_string()));
}
}
@@ -1180,13 +1175,13 @@ where
pub async fn set_group_status(&self, name: &str, enable: bool) -> Result<OffsetDateTime> {
if name.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let groups = self.cache.groups.load();
let mut gi = match groups.get(name) {
Some(gi) => gi.clone(),
None => return Err(Error::new(IamError::NoSuchGroup(name.to_string()))),
None => return Err(Error::NoSuchGroup(name.to_string())),
};
if enable {
@@ -1212,7 +1207,7 @@ where
.load()
.get(name)
.cloned()
.ok_or(Error::new(IamError::NoSuchGroup(name.to_string())))?;
.ok_or(Error::NoSuchGroup(name.to_string()))?;
Ok(GroupDesc {
name: name.to_string(),
@@ -1239,7 +1234,7 @@ where
.load()
.get(name)
.cloned()
.ok_or(Error::new(IamError::NoSuchGroup(name.to_string())))?;
.ok_or(Error::NoSuchGroup(name.to_string()))?;
let s: HashSet<&String> = HashSet::from_iter(gi.members.iter());
let d: HashSet<&String> = HashSet::from_iter(members.iter());
@@ -1265,7 +1260,7 @@ where
pub async fn remove_users_from_group(&self, group: &str, members: Vec<String>) -> Result<OffsetDateTime> {
if group.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(Error::InvalidArgument);
}
let users_cache = self.cache.users.load();
@@ -1273,10 +1268,10 @@ where
for member in members.iter() {
if let Some(u) = users_cache.get(member) {
if u.credentials.is_temp() || u.credentials.is_service_account() {
return Err(Error::new(IamError::IAMActionNotAllowed));
return Err(Error::IAMActionNotAllowed);
}
} else {
return Err(Error::new(IamError::NoSuchUser(member.to_string())));
return Err(Error::NoSuchUser(member.to_string()));
}
}
@@ -1286,10 +1281,10 @@ where
.load()
.get(group)
.cloned()
.ok_or(Error::new(IamError::NoSuchGroup(group.to_string())))?;
.ok_or(Error::NoSuchGroup(group.to_string()))?;
if members.is_empty() && !gi.members.is_empty() {
return Err(IamError::GroupNotEmpty.into());
return Err(Error::GroupNotEmpty);
}
if members.is_empty() {
@@ -1588,7 +1583,7 @@ pub fn get_token_signing_key() -> Option<String> {
pub fn extract_jwt_claims(u: &UserIdentity) -> Result<HashMap<String, Value>> {
let Some(sys_key) = get_token_signing_key() else {
return Err(Error::msg("global active sk not init"));
return Err(Error::other("global active sk not init"));
};
let keys = vec![&sys_key, &u.credentials.secret_key];
@@ -1598,7 +1593,7 @@ pub fn extract_jwt_claims(u: &UserIdentity) -> Result<HashMap<String, Value>> {
return Ok(claims);
}
}
Err(Error::msg("unable to extract claims"))
Err(Error::other("unable to extract claims"))
}
fn filter_policies(cache: &Cache, policy_name: &str, bucket_name: &str) -> (String, Policy) {
+1 -1
View File
@@ -1,7 +1,7 @@
pub mod object;
use crate::cache::Cache;
use common::error::Result;
use crate::error::Result;
use policy::{auth::UserIdentity, policy::PolicyDoc};
use serde::{de::DeserializeOwned, Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
+24 -23
View File
@@ -1,15 +1,14 @@
use super::{GroupInfo, MappedPolicy, Store, UserType};
use crate::error::{is_err_config_not_found, Error, Result};
use crate::{
cache::{Cache, CacheEntity},
error::{is_err_no_such_policy, is_err_no_such_user},
get_global_action_cred,
manager::{extract_jwt_claims, get_default_policyes},
};
use common::error::{Error, Result};
use ecstore::{
config::{
com::{delete_config, read_config, read_config_with_metadata, save_config},
error::is_err_config_not_found,
RUSTFS_CONFIG_PREFIX,
},
store::ECStore,
@@ -153,7 +152,7 @@ impl ObjectStore {
let _ = sender
.send(StringOrErr {
item: None,
err: Some(err),
err: Some(err.into()),
})
.await;
return;
@@ -213,7 +212,7 @@ impl ObjectStore {
Ok(p) => Ok(p),
Err(err) => {
if !is_err_no_such_policy(&err) {
Err(Error::msg(std::format!("load policy doc failed: {}", err)))
Err(Error::other(format!("load policy doc failed: {}", err)))
} else {
Ok(PolicyDoc::default())
}
@@ -245,7 +244,7 @@ impl ObjectStore {
Ok(res) => Ok(res),
Err(err) => {
if !is_err_no_such_user(&err) {
Err(Error::msg(std::format!("load user failed: {}", err)))
Err(Error::other(format!("load user failed: {}", err)))
} else {
Ok(UserIdentity::default())
}
@@ -272,7 +271,7 @@ impl ObjectStore {
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -296,7 +295,7 @@ impl ObjectStore {
Ok(p) => Ok(p),
Err(err) => {
if !is_err_no_such_policy(&err) {
Err(Error::msg(std::format!("load mapped policy failed: {}", err)))
Err(Error::other(format!("load mapped policy failed: {}", err)))
} else {
Ok(MappedPolicy::default())
}
@@ -369,10 +368,12 @@ impl Store for ObjectStore {
let mut data = serde_json::to_vec(&item)?;
data = Self::encrypt_data(&data)?;
save_config(self.object_api.clone(), path.as_ref(), data).await
save_config(self.object_api.clone(), path.as_ref(), data).await?;
Ok(())
}
async fn delete_iam_config(&self, path: impl AsRef<str> + Send) -> Result<()> {
delete_config(self.object_api.clone(), path.as_ref()).await
delete_config(self.object_api.clone(), path.as_ref()).await?;
Ok(())
}
async fn save_user_identity(
@@ -390,7 +391,7 @@ impl Store for ObjectStore {
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -403,7 +404,7 @@ impl Store for ObjectStore {
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchUser(name.to_owned()))
Error::NoSuchUser(name.to_owned())
} else {
err
}
@@ -412,7 +413,7 @@ impl Store for ObjectStore {
if u.credentials.is_expired() {
let _ = self.delete_iam_config(get_user_identity_path(name, user_type)).await;
let _ = self.delete_iam_config(get_mapped_policy_path(name, user_type, false)).await;
return Err(Error::new(crate::error::Error::NoSuchUser(name.to_owned())));
return Err(Error::NoSuchUser(name.to_owned()));
}
if u.credentials.access_key.is_empty() {
@@ -430,7 +431,7 @@ impl Store for ObjectStore {
let _ = self.delete_iam_config(get_mapped_policy_path(name, user_type, false)).await;
}
warn!("extract_jwt_claims failed: {}", err);
return Err(Error::new(crate::error::Error::NoSuchUser(name.to_owned())));
return Err(Error::NoSuchUser(name.to_owned()));
}
}
}
@@ -476,7 +477,7 @@ impl Store for ObjectStore {
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchUser(name.to_owned()))
Error::NoSuchUser(name.to_owned())
} else {
err
}
@@ -491,7 +492,7 @@ impl Store for ObjectStore {
async fn delete_group_info(&self, name: &str) -> Result<()> {
self.delete_iam_config(get_group_info_path(name)).await.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -501,7 +502,7 @@ impl Store for ObjectStore {
async fn load_group(&self, name: &str, m: &mut HashMap<String, GroupInfo>) -> Result<()> {
let u: GroupInfo = self.load_iam_config(get_group_info_path(name)).await.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -539,7 +540,7 @@ impl Store for ObjectStore {
async fn delete_policy_doc(&self, name: &str) -> Result<()> {
self.delete_iam_config(get_policy_doc_path(name)).await.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -552,7 +553,7 @@ impl Store for ObjectStore {
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -613,7 +614,7 @@ impl Store for ObjectStore {
.await
.map_err(|err| {
if is_err_config_not_found(&err) {
Error::new(crate::error::Error::NoSuchPolicy)
Error::NoSuchPolicy
} else {
err
}
@@ -766,7 +767,7 @@ impl Store for ObjectStore {
let name = ecstore::utils::path::dir(item);
info!("load group: {}", name);
if let Err(err) = self.load_group(&name, &mut items_cache).await {
return Err(Error::msg(std::format!("load group failed: {}", err)));
return Err(Error::other(format!("load group failed: {}", err)));
};
}
@@ -827,7 +828,7 @@ impl Store for ObjectStore {
info!("load group policy: {}", name);
if let Err(err) = self.load_mapped_policy(name, UserType::Reg, true, &mut items_cache).await {
if !is_err_no_such_policy(&err) {
return Err(Error::msg(std::format!("load group policy failed: {}", err)));
return Err(Error::other(format!("load group policy failed: {}", err)));
}
};
}
@@ -846,7 +847,7 @@ impl Store for ObjectStore {
info!("load svc user: {}", name);
if let Err(err) = self.load_user(&name, UserType::Svc, &mut items_cache).await {
if !is_err_no_such_user(&err) {
return Err(Error::msg(std::format!("load svc user failed: {}", err)));
return Err(Error::other(format!("load svc user failed: {}", err)));
}
};
}
@@ -860,7 +861,7 @@ impl Store for ObjectStore {
.await
{
if !is_err_no_such_policy(&err) {
return Err(Error::msg(std::format!("load_mapped_policy failed: {}", err)));
return Err(Error::other(format!("load_mapped_policy failed: {}", err)));
}
}
}
+25 -25
View File
@@ -1,6 +1,7 @@
use crate::error::is_err_no_such_account;
use crate::error::is_err_no_such_temp_account;
use crate::error::Error as IamError;
use crate::error::{Error, Result};
use crate::get_global_action_cred;
use crate::manager::extract_jwt_claims;
use crate::manager::get_default_policyes;
@@ -8,7 +9,6 @@ use crate::manager::IamCache;
use crate::store::MappedPolicy;
use crate::store::Store;
use crate::store::UserType;
use common::error::{Error, Result};
use ecstore::utils::crypto::base64_decode;
use ecstore::utils::crypto::base64_encode;
use madmin::AddOrUpdateUserReq;
@@ -81,7 +81,7 @@ impl<T: Store> IamSys<T> {
pub async fn delete_policy(&self, name: &str, notify: bool) -> Result<()> {
for k in get_default_policyes().keys() {
if k == name {
return Err(Error::msg("system policy can not be deleted"));
return Err(Error::other("system policy can not be deleted"));
}
}
@@ -123,11 +123,11 @@ impl<T: Store> IamSys<T> {
pub async fn get_role_policy(&self, arn_str: &str) -> Result<(ARN, String)> {
let Some(arn) = ARN::parse(arn_str).ok() else {
return Err(Error::msg("Invalid ARN"));
return Err(Error::other("Invalid ARN"));
};
let Some(policy) = self.roles_map.get(&arn) else {
return Err(Error::msg("No such role"));
return Err(Error::other("No such role"));
};
Ok((arn, policy.clone()))
@@ -157,7 +157,7 @@ impl<T: Store> IamSys<T> {
pub async fn is_temp_user(&self, name: &str) -> Result<(bool, String)> {
let Some(u) = self.store.get_user(name).await else {
return Err(IamError::NoSuchUser(name.to_string()).into());
return Err(IamError::NoSuchUser(name.to_string()));
};
if u.credentials.is_temp() {
Ok((true, u.credentials.parent_user))
@@ -167,7 +167,7 @@ impl<T: Store> IamSys<T> {
}
pub async fn is_service_account(&self, name: &str) -> Result<(bool, String)> {
let Some(u) = self.store.get_user(name).await else {
return Err(IamError::NoSuchUser(name.to_string()).into());
return Err(IamError::NoSuchUser(name.to_string()));
};
if u.credentials.is_service_account() {
@@ -193,22 +193,22 @@ impl<T: Store> IamSys<T> {
opts: NewServiceAccountOpts,
) -> Result<(Credentials, OffsetDateTime)> {
if parent_user.is_empty() {
return Err(IamError::InvalidArgument.into());
return Err(IamError::InvalidArgument);
}
if !opts.access_key.is_empty() && opts.secret_key.is_empty() {
return Err(IamError::NoSecretKeyWithAccessKey.into());
return Err(IamError::NoSecretKeyWithAccessKey);
}
if !opts.secret_key.is_empty() && opts.access_key.is_empty() {
return Err(IamError::NoAccessKeyWithSecretKey.into());
return Err(IamError::NoAccessKeyWithSecretKey);
}
if parent_user == opts.access_key {
return Err(IamError::IAMActionNotAllowed.into());
return Err(IamError::IAMActionNotAllowed);
}
if opts.expiration.is_none() {
return Err(IamError::InvalidExpiration.into());
return Err(IamError::InvalidExpiration);
}
// TODO: check allow_site_replicator_account
@@ -217,7 +217,7 @@ impl<T: Store> IamSys<T> {
policy.validate()?;
let buf = serde_json::to_vec(&policy)?;
if buf.len() > MAX_SVCSESSION_POLICY_SIZE {
return Err(IamError::PolicyTooLarge.into());
return Err(IamError::PolicyTooLarge);
}
buf
@@ -304,7 +304,7 @@ impl<T: Store> IamSys<T> {
Ok(res) => res,
Err(err) => {
if is_err_no_such_account(&err) {
return Err(IamError::NoSuchServiceAccount(access_key.to_string()).into());
return Err(IamError::NoSuchServiceAccount(access_key.to_string()));
}
return Err(err);
@@ -312,7 +312,7 @@ impl<T: Store> IamSys<T> {
};
if !sa.credentials.is_service_account() {
return Err(IamError::NoSuchServiceAccount(access_key.to_string()).into());
return Err(IamError::NoSuchServiceAccount(access_key.to_string()));
}
let op_pt = claims.get(&iam_policy_claim_name_sa());
@@ -329,7 +329,7 @@ impl<T: Store> IamSys<T> {
async fn get_account_with_claims(&self, access_key: &str) -> Result<(UserIdentity, HashMap<String, Value>)> {
let Some(acc) = self.store.get_user(access_key).await else {
return Err(IamError::NoSuchAccount(access_key.to_string()).into());
return Err(IamError::NoSuchAccount(access_key.to_string()));
};
let m = extract_jwt_claims(&acc)?;
@@ -363,7 +363,7 @@ impl<T: Store> IamSys<T> {
Ok(res) => res,
Err(err) => {
if is_err_no_such_account(&err) {
return Err(IamError::NoSuchTempAccount(access_key.to_string()).into());
return Err(IamError::NoSuchTempAccount(access_key.to_string()));
}
return Err(err);
@@ -371,7 +371,7 @@ impl<T: Store> IamSys<T> {
};
if !sa.credentials.is_temp() {
return Err(IamError::NoSuchTempAccount(access_key.to_string()).into());
return Err(IamError::NoSuchTempAccount(access_key.to_string()));
}
let op_pt = claims.get(&iam_policy_claim_name_sa());
@@ -388,11 +388,11 @@ impl<T: Store> IamSys<T> {
pub async fn get_claims_for_svc_acc(&self, access_key: &str) -> Result<HashMap<String, Value>> {
let Some(u) = self.store.get_user(access_key).await else {
return Err(IamError::NoSuchServiceAccount(access_key.to_string()).into());
return Err(IamError::NoSuchServiceAccount(access_key.to_string()));
};
if u.credentials.is_service_account() {
return Err(IamError::NoSuchServiceAccount(access_key.to_string()).into());
return Err(IamError::NoSuchServiceAccount(access_key.to_string()));
}
extract_jwt_claims(&u)
@@ -414,15 +414,15 @@ impl<T: Store> IamSys<T> {
pub async fn create_user(&self, access_key: &str, args: &AddOrUpdateUserReq) -> Result<OffsetDateTime> {
if !is_access_key_valid(access_key) {
return Err(IamError::InvalidAccessKeyLength.into());
return Err(IamError::InvalidAccessKeyLength);
}
if contains_reserved_chars(access_key) {
return Err(IamError::ContainsReservedChars.into());
return Err(IamError::ContainsReservedChars);
}
if !is_secret_key_valid(&args.secret_key) {
return Err(IamError::InvalidSecretKeyLength.into());
return Err(IamError::InvalidSecretKeyLength);
}
self.store.add_user(access_key, args).await
@@ -431,11 +431,11 @@ impl<T: Store> IamSys<T> {
pub async fn set_user_secret_key(&self, access_key: &str, secret_key: &str) -> Result<()> {
if !is_access_key_valid(access_key) {
return Err(IamError::InvalidAccessKeyLength.into());
return Err(IamError::InvalidAccessKeyLength);
}
if !is_secret_key_valid(secret_key) {
return Err(IamError::InvalidSecretKeyLength.into());
return Err(IamError::InvalidSecretKeyLength);
}
self.store.update_user_secret_key(access_key, secret_key).await
@@ -467,7 +467,7 @@ impl<T: Store> IamSys<T> {
pub async fn add_users_to_group(&self, group: &str, users: Vec<String>) -> Result<OffsetDateTime> {
if contains_reserved_chars(group) {
return Err(IamError::GroupNameContainsReservedChars.into());
return Err(IamError::GroupNameContainsReservedChars);
}
self.store.add_users_to_group(group, users).await
// TODO: notification
+5 -5
View File
@@ -1,7 +1,7 @@
use common::error::{Error, Result};
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header};
use rand::{Rng, RngCore};
use serde::{de::DeserializeOwned, Serialize};
use std::io::{Error, Result};
pub fn gen_access_key(length: usize) -> Result<String> {
const ALPHA_NUMERIC_TABLE: [char; 36] = [
@@ -10,7 +10,7 @@ pub fn gen_access_key(length: usize) -> Result<String> {
];
if length < 3 {
return Err(Error::msg("access key length is too short"));
return Err(Error::other("access key length is too short"));
}
let mut result = String::with_capacity(length);
@@ -27,7 +27,7 @@ pub fn gen_secret_key(length: usize) -> Result<String> {
use base64_simd::URL_SAFE_NO_PAD;
if length < 8 {
return Err(Error::msg("secret key length is too short"));
return Err(Error::other("secret key length is too short"));
}
let mut rng = rand::thread_rng();
@@ -40,7 +40,7 @@ pub fn gen_secret_key(length: usize) -> Result<String> {
Ok(key_str)
}
pub fn generate_jwt<T: Serialize>(claims: &T, secret: &str) -> Result<String, jsonwebtoken::errors::Error> {
pub fn generate_jwt<T: Serialize>(claims: &T, secret: &str) -> std::result::Result<String, jsonwebtoken::errors::Error> {
let header = Header::new(Algorithm::HS512);
jsonwebtoken::encode(&header, &claims, &EncodingKey::from_secret(secret.as_bytes()))
}
@@ -48,7 +48,7 @@ pub fn generate_jwt<T: Serialize>(claims: &T, secret: &str) -> Result<String, js
pub fn extract_claims<T: DeserializeOwned>(
token: &str,
secret: &str,
) -> Result<jsonwebtoken::TokenData<T>, jsonwebtoken::errors::Error> {
) -> std::result::Result<jsonwebtoken::TokenData<T>, jsonwebtoken::errors::Error> {
jsonwebtoken::decode::<T>(
token,
&DecodingKey::from_secret(secret.as_bytes()),
+1 -1
View File
@@ -29,7 +29,7 @@ tracing.workspace = true
madmin.workspace = true
lazy_static.workspace = true
regex = "1.11.1"
common.workspace = true
[dev-dependencies]
test-case.workspace = true
+9 -9
View File
@@ -1,4 +1,4 @@
use common::error::{Error, Result};
use crate::error::{Error, Result};
use regex::Regex;
const ARN_PREFIX_ARN: &str = "arn";
@@ -19,7 +19,7 @@ impl ARN {
pub fn new_iam_role_arn(resource_id: &str, server_region: &str) -> Result<Self> {
let valid_resource_id_regex = Regex::new(r"^[A-Za-z0-9_/\.-]+$")?;
if !valid_resource_id_regex.is_match(resource_id) {
return Err(Error::msg("ARN resource ID invalid"));
return Err(Error::other("ARN resource ID invalid"));
}
Ok(ARN {
partition: ARN_PARTITION_RUSTFS.to_string(),
@@ -33,33 +33,33 @@ impl ARN {
pub fn parse(arn_str: &str) -> Result<Self> {
let ps: Vec<&str> = arn_str.split(':').collect();
if ps.len() != 6 || ps[0] != ARN_PREFIX_ARN {
return Err(Error::msg("ARN format invalid"));
return Err(Error::other("ARN format invalid"));
}
if ps[1] != ARN_PARTITION_RUSTFS {
return Err(Error::msg("ARN partition invalid"));
return Err(Error::other("ARN partition invalid"));
}
if ps[2] != ARN_SERVICE_IAM {
return Err(Error::msg("ARN service invalid"));
return Err(Error::other("ARN service invalid"));
}
if !ps[4].is_empty() {
return Err(Error::msg("ARN account-id invalid"));
return Err(Error::other("ARN account-id invalid"));
}
let res: Vec<&str> = ps[5].splitn(2, '/').collect();
if res.len() != 2 {
return Err(Error::msg("ARN resource invalid"));
return Err(Error::other("ARN resource invalid"));
}
if res[0] != ARN_RESOURCE_TYPE_ROLE {
return Err(Error::msg("ARN resource type invalid"));
return Err(Error::other("ARN resource type invalid"));
}
let valid_resource_id_regex = Regex::new(r"^[A-Za-z0-9_/\.-]+$")?;
if !valid_resource_id_regex.is_match(res[1]) {
return Err(Error::msg("ARN resource ID invalid"));
return Err(Error::other("ARN resource ID invalid"));
}
Ok(ARN {
+22 -22
View File
@@ -1,8 +1,8 @@
use crate::error::Error as IamError;
use crate::error::{Error, Result};
use crate::policy::{iam_policy_claim_name_sa, Policy, Validator, INHERITED_POLICY_TYPE};
use crate::utils;
use crate::utils::extract_claims;
use common::error::{Error, Result};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
@@ -54,41 +54,41 @@ pub fn is_secret_key_valid(secret_key: &str) -> bool {
// fn try_from(value: &str) -> Result<Self, Self::Error> {
// let mut elem = value.trim().splitn(2, '=');
// let (Some(h), Some(cred_elems)) = (elem.next(), elem.next()) else {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// };
// if h != "Credential" {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// }
// let mut cred_elems = cred_elems.trim().rsplitn(5, '/');
// let Some(request) = cred_elems.next() else {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// };
// let Some(service) = cred_elems.next() else {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// };
// let Some(region) = cred_elems.next() else {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// };
// let Some(date) = cred_elems.next() else {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// };
// let Some(ak) = cred_elems.next() else {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// };
// if ak.len() < 3 {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// }
// if request != "aws4_request" {
// return Err(Error::new(IamError::ErrCredMalformed));
// return Err(IamError::ErrCredMalformed));
// }
// Ok(CredentialHeader {
@@ -98,7 +98,7 @@ pub fn is_secret_key_valid(secret_key: &str) -> bool {
// const FORMATTER: LazyCell<Vec<BorrowedFormatItem<'static>>> =
// LazyCell::new(|| time::format_description::parse("[year][month][day]").unwrap());
// Date::parse(date, &FORMATTER).map_err(|_| Error::new(IamError::ErrCredMalformed))?
// Date::parse(date, &FORMATTER).map_err(|_| IamError::ErrCredMalformed))?
// },
// region: region.to_owned(),
// service: service.try_into()?,
@@ -199,11 +199,11 @@ pub fn create_new_credentials_with_metadata(
token_secret: &str,
) -> Result<Credentials> {
if ak.len() < ACCESS_KEY_MIN_LEN || ak.len() > ACCESS_KEY_MAX_LEN {
return Err(Error::new(IamError::InvalidAccessKeyLength));
return Err(IamError::InvalidAccessKeyLength);
}
if sk.len() < SECRET_KEY_MIN_LEN || sk.len() > SECRET_KEY_MAX_LEN {
return Err(Error::new(IamError::InvalidAccessKeyLength));
return Err(IamError::InvalidAccessKeyLength);
}
if token_secret.is_empty() {
@@ -326,23 +326,23 @@ impl CredentialsBuilder {
impl TryFrom<CredentialsBuilder> for Credentials {
type Error = Error;
fn try_from(mut value: CredentialsBuilder) -> Result<Self, Self::Error> {
fn try_from(mut value: CredentialsBuilder) -> std::result::Result<Self, Self::Error> {
if value.parent_user.is_empty() {
return Err(Error::new(IamError::InvalidArgument));
return Err(IamError::InvalidArgument);
}
if (value.access_key.is_empty() && !value.secret_key.is_empty())
|| (!value.access_key.is_empty() && value.secret_key.is_empty())
{
return Err(Error::msg("Either ak or sk is empty"));
return Err(Error::other("Either ak or sk is empty"));
}
if value.parent_user == value.access_key.as_str() {
return Err(Error::new(IamError::InvalidArgument));
return Err(IamError::InvalidArgument);
}
if value.access_key == "site-replicator-0" && !value.allow_site_replicator_account {
return Err(Error::new(IamError::InvalidArgument));
return Err(IamError::InvalidArgument);
}
let mut claim = serde_json::json!({
@@ -351,9 +351,9 @@ impl TryFrom<CredentialsBuilder> for Credentials {
if let Some(p) = value.session_policy {
p.is_valid()?;
let policy_buf = serde_json::to_vec(&p).map_err(|_| Error::new(IamError::InvalidArgument))?;
let policy_buf = serde_json::to_vec(&p).map_err(|_| IamError::InvalidArgument)?;
if policy_buf.len() > 4096 {
return Err(Error::msg("session policy is too large"));
return Err(Error::other("session policy is too large"));
}
claim["sessionPolicy"] = serde_json::json!(base64_simd::STANDARD.encode_to_string(&policy_buf));
claim["sa-policy"] = serde_json::json!("embedded-policy");
@@ -390,8 +390,8 @@ impl TryFrom<CredentialsBuilder> for Credentials {
};
if !value.secret_key.is_empty() {
let session_token =
crypto::jwt_encode(value.access_key.as_bytes(), &claim).map_err(|_| Error::msg("session policy is too large"))?;
let session_token = crypto::jwt_encode(value.access_key.as_bytes(), &claim)
.map_err(|_| Error::other("session policy is too large"))?;
cred.session_token = session_token;
// cred.expiration = Some(
// OffsetDateTime::from_unix_timestamp(
+57 -40
View File
@@ -1,13 +1,12 @@
use crate::policy;
pub type Result<T> = core::result::Result<T, Error>;
#[derive(thiserror::Error, Debug)]
pub enum Error {
#[error(transparent)]
PolicyError(#[from] policy::Error),
#[error("ecsotre error: {0}")]
EcstoreError(common::error::Error),
#[error("{0}")]
StringError(String),
@@ -66,7 +65,7 @@ pub enum Error {
GroupNameContainsReservedChars,
#[error("jwt err {0}")]
JWTError(jsonwebtoken::errors::Error),
JWTError(#[from] jsonwebtoken::errors::Error),
#[error("no access key")]
NoAccessKey,
@@ -90,56 +89,74 @@ pub enum Error {
#[error("policy too large")]
PolicyTooLarge,
#[error("io error: {0}")]
Io(std::io::Error),
}
impl Error {
pub fn other<E>(error: E) -> Self
where
E: Into<Box<dyn std::error::Error + Send + Sync>>,
{
Error::Io(std::io::Error::other(error))
}
}
impl From<std::io::Error> for Error {
fn from(e: std::io::Error) -> Self {
Error::Io(e)
}
}
impl From<time::error::ComponentRange> for Error {
fn from(e: time::error::ComponentRange) -> Self {
Error::other(e)
}
}
impl From<serde_json::Error> for Error {
fn from(e: serde_json::Error) -> Self {
Error::other(e)
}
}
// impl From<jsonwebtoken::errors::Error> for Error {
// fn from(e: jsonwebtoken::errors::Error) -> Self {
// Error::JWTError(e)
// }
// }
impl From<regex::Error> for Error {
fn from(e: regex::Error) -> Self {
Error::other(e)
}
}
// pub fn is_err_no_such_user(e: &Error) -> bool {
// matches!(e, Error::NoSuchUser(_))
// }
pub fn is_err_no_such_policy(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchPolicy)
} else {
false
}
pub fn is_err_no_such_policy(err: &Error) -> bool {
matches!(err, Error::NoSuchPolicy)
}
pub fn is_err_no_such_user(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchUser(_))
} else {
false
}
pub fn is_err_no_such_user(err: &Error) -> bool {
matches!(err, Error::NoSuchUser(_))
}
pub fn is_err_no_such_account(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchAccount(_))
} else {
false
}
pub fn is_err_no_such_account(err: &Error) -> bool {
matches!(err, Error::NoSuchAccount(_))
}
pub fn is_err_no_such_temp_account(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchTempAccount(_))
} else {
false
}
pub fn is_err_no_such_temp_account(err: &Error) -> bool {
matches!(err, Error::NoSuchTempAccount(_))
}
pub fn is_err_no_such_group(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchGroup(_))
} else {
false
}
pub fn is_err_no_such_group(err: &Error) -> bool {
matches!(err, Error::NoSuchGroup(_))
}
pub fn is_err_no_such_service_account(err: &common::error::Error) -> bool {
if let Some(e) = err.downcast_ref::<Error>() {
matches!(e, Error::NoSuchServiceAccount(_))
} else {
false
}
pub fn is_err_no_such_service_account(err: &Error) -> bool {
matches!(err, Error::NoSuchServiceAccount(_))
}
+2 -2
View File
@@ -1,4 +1,4 @@
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use std::{collections::HashSet, ops::Deref};
use strum::{EnumString, IntoStaticStr};
@@ -84,7 +84,7 @@ impl Action {
impl TryFrom<&str> for Action {
type Error = Error;
fn try_from(value: &str) -> Result<Self, Self::Error> {
fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
if value.starts_with(Self::S3_PREFIX) {
Ok(Self::S3Action(
S3Action::try_from(value).map_err(|_| IamError::InvalidAction(value.into()))?,
+1 -1
View File
@@ -1,4 +1,4 @@
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use strum::{EnumString, IntoStaticStr};
+1 -1
View File
@@ -1,6 +1,6 @@
use super::key_name::KeyName;
use crate::error::Error;
use crate::policy::{Error as PolicyError, Validator};
use common::error::Error;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
+1 -1
View File
@@ -1,4 +1,4 @@
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use std::ops::Deref;
+2 -2
View File
@@ -1,5 +1,5 @@
use super::{action::Action, statement::BPStatement, Effect, Error as IamError, Statement, ID};
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{HashMap, HashSet};
@@ -449,7 +449,7 @@ pub mod default {
#[cfg(test)]
mod test {
use super::*;
use common::error::Result;
use crate::error::Result;
#[tokio::test]
async fn test_parse_policy() -> Result<()> {
+2 -2
View File
@@ -1,5 +1,5 @@
use super::{utils::wildcard, Validator};
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
@@ -25,7 +25,7 @@ impl Validator for Principal {
type Error = Error;
fn is_valid(&self) -> Result<()> {
if self.aws.is_empty() {
return Err(Error::msg("Principal is empty"));
return Err(Error::other("Principal is empty"));
}
Ok(())
}
+5 -5
View File
@@ -1,4 +1,4 @@
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
use std::{
collections::{HashMap, HashSet},
@@ -101,7 +101,7 @@ impl Resource {
impl TryFrom<&str> for Resource {
type Error = Error;
fn try_from(value: &str) -> Result<Self, Self::Error> {
fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
let resource = if value.starts_with(Self::S3_PREFIX) {
Resource::S3(value.strip_prefix(Self::S3_PREFIX).unwrap().into())
} else {
@@ -115,7 +115,7 @@ impl TryFrom<&str> for Resource {
impl Validator for Resource {
type Error = Error;
fn is_valid(&self) -> Result<(), Error> {
fn is_valid(&self) -> std::result::Result<(), Error> {
match self {
Self::S3(pattern) => {
if pattern.is_empty() || pattern.starts_with('/') {
@@ -139,7 +139,7 @@ impl Validator for Resource {
}
impl Serialize for Resource {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
@@ -151,7 +151,7 @@ impl Serialize for Resource {
}
impl<'de> Deserialize<'de> for Resource {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
+1 -1
View File
@@ -2,7 +2,7 @@ use super::{
action::Action, ActionSet, Args, BucketPolicyArgs, Effect, Error as IamError, Functions, Principal, ResourceSet, Validator,
ID,
};
use common::error::{Error, Result};
use crate::error::{Error, Result};
use serde::{Deserialize, Serialize};
#[derive(Serialize, Deserialize, Clone, Default, Debug)]
+5 -5
View File
@@ -1,7 +1,7 @@
use common::error::{Error, Result};
use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header};
use rand::{Rng, RngCore};
use serde::{de::DeserializeOwned, Serialize};
use std::io::{Error, Result};
pub fn gen_access_key(length: usize) -> Result<String> {
const ALPHA_NUMERIC_TABLE: [char; 36] = [
@@ -10,7 +10,7 @@ pub fn gen_access_key(length: usize) -> Result<String> {
];
if length < 3 {
return Err(Error::msg("access key length is too short"));
return Err(Error::other("access key length is too short"));
}
let mut result = String::with_capacity(length);
@@ -27,7 +27,7 @@ pub fn gen_secret_key(length: usize) -> Result<String> {
use base64_simd::URL_SAFE_NO_PAD;
if length < 8 {
return Err(Error::msg("secret key length is too short"));
return Err(Error::other("secret key length is too short"));
}
let mut rng = rand::thread_rng();
@@ -40,7 +40,7 @@ pub fn gen_secret_key(length: usize) -> Result<String> {
Ok(key_str)
}
pub fn generate_jwt<T: Serialize>(claims: &T, secret: &str) -> Result<String, jsonwebtoken::errors::Error> {
pub fn generate_jwt<T: Serialize>(claims: &T, secret: &str) -> std::result::Result<String, jsonwebtoken::errors::Error> {
let header = Header::new(Algorithm::HS512);
jsonwebtoken::encode(&header, &claims, &EncodingKey::from_secret(secret.as_bytes()))
}
@@ -48,7 +48,7 @@ pub fn generate_jwt<T: Serialize>(claims: &T, secret: &str) -> Result<String, js
pub fn extract_claims<T: DeserializeOwned>(
token: &str,
secret: &str,
) -> Result<jsonwebtoken::TokenData<T>, jsonwebtoken::errors::Error> {
) -> std::result::Result<jsonwebtoken::TokenData<T>, jsonwebtoken::errors::Error> {
jsonwebtoken::decode::<T>(
token,
&DecodingKey::from_secret(secret.as_bytes()),
+2
View File
@@ -86,6 +86,8 @@ tower-http = { workspace = true, features = [
"cors",
] }
uuid = { workspace = true }
rustfs-filemeta.workspace = true
thiserror.workspace = true
[target.'cfg(target_os = "linux")'.dependencies]
libsystemd.workspace = true
+6 -6
View File
@@ -3,14 +3,14 @@ use crate::auth::check_key_valid;
use crate::auth::get_condition_values;
use crate::auth::get_session_token;
use bytes::Bytes;
use common::error::Error as ec_Error;
use ecstore::admin_server_info::get_server_info;
use ecstore::bucket::versioning_sys::BucketVersioningSys;
use ecstore::error::StorageError;
use ecstore::global::GLOBAL_ALlHealState;
use ecstore::heal::data_usage::load_data_usage_from_backend;
use ecstore::heal::heal_commands::HealOpts;
use ecstore::heal::heal_ops::new_heal_sequence;
use ecstore::metrics_realtime::{collect_local_metrics, CollectMetricsOpts, MetricType};
use ecstore::metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics};
use ecstore::new_object_layer_fn;
use ecstore::peer::is_reserved_or_invalid_bucket;
use ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free};
@@ -26,14 +26,14 @@ use iam::store::MappedPolicy;
use madmin::metrics::RealtimeMetrics;
use madmin::utils::parse_duration;
use matchit::Params;
use policy::policy::Args;
use policy::policy::BucketPolicy;
use policy::policy::action::Action;
use policy::policy::action::S3Action;
use policy::policy::default::DEFAULT_POLICIES;
use policy::policy::Args;
use policy::policy::BucketPolicy;
use s3s::header::CONTENT_TYPE;
use s3s::stream::{ByteStream, DynByteStream};
use s3s::{s3_error, Body, S3Error, S3Request, S3Response, S3Result};
use s3s::{Body, S3Error, S3Request, S3Response, S3Result, s3_error};
use s3s::{S3ErrorCode, StdError};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
@@ -656,7 +656,7 @@ impl Operation for HealHandler {
#[derive(Default)]
struct HealResp {
resp_bytes: Vec<u8>,
_api_err: Option<ec_Error>,
_api_err: Option<StorageError>,
_err_body: String,
}
+4 -12
View File
@@ -1,7 +1,7 @@
use ecstore::{new_object_layer_fn, GLOBAL_Endpoints};
use ecstore::{GLOBAL_Endpoints, new_object_layer_fn};
use http::{HeaderMap, StatusCode};
use matchit::Params;
use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result};
use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::Deserialize;
use serde_urlencoded::from_bytes;
use tokio::sync::broadcast;
@@ -88,11 +88,7 @@ impl Operation for StatusPool {
let has_idx = {
if is_byid {
let a = query.pool.parse::<usize>().unwrap_or_default();
if a < endpoints.as_ref().len() {
Some(a)
} else {
None
}
if a < endpoints.as_ref().len() { Some(a) } else { None }
} else {
endpoints.get_pool_idx(&query.pool)
}
@@ -234,11 +230,7 @@ impl Operation for CancelDecommission {
let has_idx = {
if is_byid {
let a = query.pool.parse::<usize>().unwrap_or_default();
if a < endpoints.as_ref().len() {
Some(a)
} else {
None
}
if a < endpoints.as_ref().len() { Some(a) } else { None }
} else {
endpoints.get_pool_idx(&query.pool)
}
+4 -4
View File
@@ -1,14 +1,14 @@
use ecstore::{
config::error::is_err_config_not_found,
StorageAPI,
error::StorageError,
new_object_layer_fn,
notification_sys::get_global_notification_sys,
rebalance::{DiskStat, RebalSaveOpt},
store_api::BucketOptions,
StorageAPI,
};
use http::{HeaderMap, StatusCode};
use matchit::Params;
use s3s::{header::CONTENT_TYPE, s3_error, Body, S3Request, S3Response, S3Result};
use s3s::{Body, S3Request, S3Response, S3Result, header::CONTENT_TYPE, s3_error};
use serde::{Deserialize, Serialize};
use std::time::{Duration, SystemTime};
use tracing::warn;
@@ -133,7 +133,7 @@ impl Operation for RebalanceStatus {
let mut meta = RebalanceMeta::new();
if let Err(err) = meta.load(store.pools[0].clone()).await {
if is_err_config_not_found(&err) {
if err == StorageError::ConfigNotFound {
return Err(s3_error!(NoSuchResource, "Pool rebalance is not started"));
}
+1687
View File
File diff suppressed because it is too large Load Diff
+69 -289
View File
@@ -6,26 +6,25 @@ use ecstore::{
bucket::{metadata::load_bucket_metadata, metadata_sys},
disk::{
DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts,
error::DiskError,
},
error::StorageError,
heal::{
data_usage_cache::DataUsageCache,
heal_commands::{get_local_background_heal_status, HealOpts},
heal_commands::{HealOpts, get_local_background_heal_status},
},
metrics_realtime::{collect_local_metrics, CollectMetricsOpts, MetricType},
metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics},
new_object_layer_fn,
peer::{LocalPeerS3Client, PeerS3Client},
store::{all_local_disk_path, find_local_disk},
store_api::{BucketOptions, DeleteBucketOptions, FileInfo, MakeBucketOptions, StorageAPI},
utils::err_to_proto_err,
store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI},
};
use futures::{Stream, StreamExt};
use futures_util::future::join_all;
use lock::{lock_args::LockArgs, Locker, GLOBAL_LOCAL_SERVER};
use lock::{GLOBAL_LOCAL_SERVER, Locker, lock_args::LockArgs};
use common::globals::GLOBAL_Local_Node_Name;
use ecstore::disk::error::is_err_eof;
use ecstore::metacache::writer::MetacacheReader;
use madmin::health::{
get_cpus, get_mem_info, get_os_info, get_partitions, get_proc_info, get_sys_config, get_sys_errors, get_sys_services,
};
@@ -35,6 +34,7 @@ use protos::{
proto_gen::node_service::{node_service_server::NodeService as Node, *},
};
use rmp_serde::{Deserializer, Serializer};
use rustfs_filemeta::{FileInfo, MetacacheReader};
use serde::{Deserialize, Serialize};
use tokio::spawn;
use tokio::sync::mpsc;
@@ -121,11 +121,8 @@ impl Node for NodeService {
Err(err) => {
return Ok(tonic::Response::new(HealBucketResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
&format!("decode HealOpts failed: {}", err),
)),
}))
error: Some(DiskError::other(format!("decode HealOpts failed: {}", err)).into()),
}));
}
};
@@ -137,7 +134,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(HealBucketResponse {
success: false,
error: Some(err_to_proto_err(&err, &format!("heal bucket failed: {}", err))),
error: Some(err.into()),
})),
}
}
@@ -152,11 +149,8 @@ impl Node for NodeService {
return Ok(tonic::Response::new(ListBucketResponse {
success: false,
bucket_infos: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
&format!("decode BucketOptions failed: {}", err),
)),
}))
error: Some(DiskError::other(format!("decode BucketOptions failed: {}", err)).into()),
}));
}
};
match self.local_peer.list_bucket(&options).await {
@@ -175,7 +169,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(ListBucketResponse {
success: false,
bucket_infos: Vec::new(),
error: Some(err_to_proto_err(&err, &format!("list bucket failed: {}", err))),
error: Some(err.into()),
})),
}
}
@@ -189,11 +183,8 @@ impl Node for NodeService {
Err(err) => {
return Ok(tonic::Response::new(MakeBucketResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
&format!("decode MakeBucketOptions failed: {}", err),
)),
}))
error: Some(DiskError::other(format!("decode MakeBucketOptions failed: {}", err)).into()),
}));
}
};
match self.local_peer.make_bucket(&request.name, &options).await {
@@ -203,7 +194,7 @@ impl Node for NodeService {
})),
Err(err) => Ok(tonic::Response::new(MakeBucketResponse {
success: false,
error: Some(err_to_proto_err(&err, &format!("make bucket failed: {}", err))),
error: Some(err.into()),
})),
}
}
@@ -218,11 +209,8 @@ impl Node for NodeService {
return Ok(tonic::Response::new(GetBucketInfoResponse {
success: false,
bucket_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
&format!("decode BucketOptions failed: {}", err),
)),
}))
error: Some(DiskError::other(format!("decode BucketOptions failed: {}", err)).into()),
}));
}
};
match self.local_peer.get_bucket_info(&request.bucket, &options).await {
@@ -233,10 +221,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(GetBucketInfoResponse {
success: false,
bucket_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
}));
}
};
@@ -250,7 +235,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(GetBucketInfoResponse {
success: false,
bucket_info: String::new(),
error: Some(err_to_proto_err(&err, &format!("get bucket info failed: {}", err))),
error: Some(err.into()),
})),
}
}
@@ -276,7 +261,7 @@ impl Node for NodeService {
})),
Err(err) => Ok(tonic::Response::new(DeleteBucketResponse {
success: false,
error: Some(err_to_proto_err(&err, &format!("delete bucket failed: {}", err))),
error: Some(err.into()),
})),
}
}
@@ -302,10 +287,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(ReadAllResponse {
success: false,
data: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -326,10 +308,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(WriteAllResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -342,14 +321,7 @@ impl Node for NodeService {
Err(err) => {
return Ok(tonic::Response::new(DeleteResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode DeleteOptions failed: {}", err),
)),
error: Some(DiskError::other(format!("decode DeleteOptions failed: {}", err)).into()),
}));
}
};
@@ -366,10 +338,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(DeleteResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -383,14 +352,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(VerifyFileResponse {
success: false,
check_parts_resp: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfo failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfo failed: {}", err)).into()),
}));
}
};
@@ -402,10 +364,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(VerifyFileResponse {
success: false,
check_parts_resp: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
}));
}
};
@@ -425,10 +384,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(VerifyFileResponse {
success: false,
check_parts_resp: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -442,14 +398,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(CheckPartsResponse {
success: false,
check_parts_resp: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfo failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfo failed: {}", err)).into()),
}));
}
};
@@ -461,10 +410,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(CheckPartsResponse {
success: false,
check_parts_resp: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
}));
}
};
@@ -484,10 +430,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(CheckPartsResponse {
success: false,
check_parts_resp: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -517,10 +460,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(RenamePartResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -544,10 +484,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(RenameFileResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -812,10 +749,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(ListDirResponse {
success: false,
volumes: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -870,7 +804,7 @@ impl Node for NodeService {
}
}
Err(err) => {
if is_err_eof(&err) {
if rustfs_filemeta::is_io_eof(&err) {
break;
}
@@ -899,14 +833,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(RenameDataResponse {
success: false,
rename_data_resp: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfo failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfo failed: {}", err)).into()),
}));
}
};
@@ -921,10 +848,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(RenameDataResponse {
success: false,
rename_data_resp: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
}));
}
};
@@ -944,10 +868,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(RenameDataResponse {
success: false,
rename_data_resp: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -968,10 +889,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(MakeVolumesResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -992,10 +910,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(MakeVolumeResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1025,10 +940,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(ListVolumesResponse {
success: false,
volume_infos: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1046,10 +958,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(StatVolumeResponse {
success: false,
volume_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
})),
},
Err(err) => Ok(tonic::Response::new(StatVolumeResponse {
@@ -1062,10 +971,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(StatVolumeResponse {
success: false,
volume_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1086,10 +992,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(DeletePathsResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1102,14 +1005,7 @@ impl Node for NodeService {
Err(err) => {
return Ok(tonic::Response::new(UpdateMetadataResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfo failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfo failed: {}", err)).into()),
}));
}
};
@@ -1118,14 +1014,7 @@ impl Node for NodeService {
Err(err) => {
return Ok(tonic::Response::new(UpdateMetadataResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode UpdateMetadataOpts failed: {}", err),
)),
error: Some(DiskError::other(format!("decode UpdateMetadataOpts failed: {}", err)).into()),
}));
}
};
@@ -1143,10 +1032,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(UpdateMetadataResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1159,14 +1045,7 @@ impl Node for NodeService {
Err(err) => {
return Ok(tonic::Response::new(WriteMetadataResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfo failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfo failed: {}", err)).into()),
}));
}
};
@@ -1183,10 +1062,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(WriteMetadataResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1200,14 +1076,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(ReadVersionResponse {
success: false,
file_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode ReadOptions failed: {}", err),
)),
error: Some(DiskError::other(format!("decode ReadOptions failed: {}", err)).into()),
}));
}
};
@@ -1224,10 +1093,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(ReadVersionResponse {
success: false,
file_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
})),
},
Err(err) => Ok(tonic::Response::new(ReadVersionResponse {
@@ -1240,10 +1106,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(ReadVersionResponse {
success: false,
file_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1261,10 +1124,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(ReadXlResponse {
success: false,
raw_file_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
})),
},
Err(err) => Ok(tonic::Response::new(ReadXlResponse {
@@ -1277,10 +1137,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(ReadXlResponse {
success: false,
raw_file_info: String::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1294,14 +1151,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(DeleteVersionResponse {
success: false,
raw_file_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfo failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfo failed: {}", err)).into()),
}));
}
};
@@ -1311,14 +1161,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(DeleteVersionResponse {
success: false,
raw_file_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode DeleteOptions failed: {}", err),
)),
error: Some(DiskError::other(format!("decode DeleteOptions failed: {}", err)).into()),
}));
}
};
@@ -1335,10 +1178,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(DeleteVersionResponse {
success: false,
raw_file_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
})),
},
Err(err) => Ok(tonic::Response::new(DeleteVersionResponse {
@@ -1351,10 +1191,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(DeleteVersionResponse {
success: false,
raw_file_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1370,14 +1207,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(DeleteVersionsResponse {
success: false,
errors: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode FileInfoVersions failed: {}", err),
)),
error: Some(DiskError::other(format!("decode FileInfoVersions failed: {}", err)).into()),
}));
}
};
@@ -1388,14 +1218,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(DeleteVersionsResponse {
success: false,
errors: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode DeleteOptions failed: {}", err),
)),
error: Some(DiskError::other(format!("decode DeleteOptions failed: {}", err)).into()),
}));
}
};
@@ -1425,10 +1248,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(DeleteVersionsResponse {
success: false,
errors: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1442,14 +1262,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(ReadMultipleResponse {
success: false,
read_multiple_resps: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode ReadMultipleReq failed: {}", err),
)),
error: Some(DiskError::other(format!("decode ReadMultipleReq failed: {}", err)).into()),
}));
}
};
@@ -1476,10 +1289,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(ReadMultipleResponse {
success: false,
read_multiple_resps: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1500,10 +1310,7 @@ impl Node for NodeService {
} else {
Ok(tonic::Response::new(DeleteVolumeResponse {
success: false,
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1517,14 +1324,7 @@ impl Node for NodeService {
return Ok(tonic::Response::new(DiskInfoResponse {
success: false,
disk_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode DiskInfoOptions failed: {}", err),
)),
error: Some(DiskError::other(format!("decode DiskInfoOptions failed: {}", err)).into()),
}));
}
};
@@ -1538,10 +1338,7 @@ impl Node for NodeService {
Err(err) => Ok(tonic::Response::new(DiskInfoResponse {
success: false,
disk_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::from_string("encode data failed"),
&format!("encode data failed: {}", err),
)),
error: Some(DiskError::other(format!("encode data failed: {}", err)).into()),
})),
},
Err(err) => Ok(tonic::Response::new(DiskInfoResponse {
@@ -1554,10 +1351,7 @@ impl Node for NodeService {
Ok(tonic::Response::new(DiskInfoResponse {
success: false,
disk_info: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
}
}
@@ -1580,14 +1374,7 @@ impl Node for NodeService {
success: false,
update: "".to_string(),
data_usage_cache: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode DataUsageCache failed: {}", err),
)),
error: Some(DiskError::other(format!("decode DataUsageCache failed: {}", err)).into()),
}))
.await
.expect("working rx");
@@ -1645,14 +1432,7 @@ impl Node for NodeService {
success: false,
update: "".to_string(),
data_usage_cache: "".to_string(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
"can not find disk",
)),
error: Some(DiskError::other("can not find disk".to_string()).into()),
}))
.await
.expect("working rx");
+13 -9
View File
@@ -2,6 +2,7 @@ mod admin;
mod auth;
mod config;
mod console;
mod error;
mod event;
mod grpc;
pub mod license;
@@ -11,9 +12,9 @@ mod service;
mod storage;
use crate::auth::IAMAuth;
use crate::console::{init_console_cfg, CONSOLE_CONFIG};
use crate::console::{CONSOLE_CONFIG, init_console_cfg};
// Ensure the correct path for parse_license is imported
use crate::server::{wait_for_shutdown, ServiceState, ServiceStateManager, ShutdownSignal, SHUTDOWN_TIMEOUT};
use crate::server::{SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, wait_for_shutdown};
use bytes::Bytes;
use chrono::Datelike;
use clap::Parser;
@@ -21,18 +22,18 @@ use common::{
error::{Error, Result},
globals::set_global_addr,
};
use ecstore::StorageAPI;
use ecstore::bucket::metadata_sys::init_bucket_metadata_sys;
use ecstore::config as ecconfig;
use ecstore::config::GLOBAL_ConfigSys;
use ecstore::heal::background_heal_ops::init_auto_heal;
use ecstore::store_api::BucketOptions;
use ecstore::utils::net;
use ecstore::StorageAPI;
use ecstore::{
endpoints::EndpointServerPools,
heal::data_scanner::init_data_scanner,
set_global_endpoints,
store::{init_local_disks, ECStore},
store::{ECStore, init_local_disks},
update_erasure_type,
};
use ecstore::{global::set_global_rustfs_port, notification_sys::new_global_notification_sys};
@@ -48,7 +49,7 @@ use iam::init_iam_sys;
use license::init_license;
use protos::proto_gen::node_service::node_service_server::NodeServiceServer;
use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY};
use rustfs_obs::{init_obs, set_global_guard, SystemObserver};
use rustfs_obs::{SystemObserver, init_obs, set_global_guard};
use rustls::ServerConfig;
use s3s::{host::MultiDomain, service::S3ServiceBuilder};
use service::hybrid;
@@ -57,13 +58,13 @@ use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpListener;
use tokio::signal::unix::{signal, SignalKind};
use tokio::signal::unix::{SignalKind, signal};
use tokio_rustls::TlsAcceptor;
use tonic::{metadata::MetadataValue, Request, Status};
use tonic::{Request, Status, metadata::MetadataValue};
use tower_http::cors::CorsLayer;
use tower_http::trace::TraceLayer;
use tracing::{Span, instrument};
use tracing::{debug, error, info, warn};
use tracing::{instrument, Span};
const MI_B: usize = 1024 * 1024;
@@ -163,7 +164,10 @@ async fn run(opt: config::Opt) -> Result<()> {
info!(" RootUser: {}", opt.access_key.clone());
info!(" RootPass: {}", opt.secret_key.clone());
if DEFAULT_ACCESS_KEY.eq(&opt.access_key) && DEFAULT_SECRET_KEY.eq(&opt.secret_key) {
warn!("Detected default credentials '{}:{}', we recommend that you change these values with 'RUSTFS_ACCESS_KEY' and 'RUSTFS_SECRET_KEY' environment variables", DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY);
warn!(
"Detected default credentials '{}:{}', we recommend that you change these values with 'RUSTFS_ACCESS_KEY' and 'RUSTFS_SECRET_KEY' environment variables",
DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY
);
}
for (i, eps) in endpoint_pools.as_ref().iter().enumerate() {
+4 -6
View File
@@ -7,7 +7,7 @@ use policy::auth;
use policy::policy::action::{Action, S3Action};
use policy::policy::{Args, BucketPolicyArgs};
use s3s::access::{S3Access, S3AccessContext};
use s3s::{dto::*, s3_error, S3Error, S3ErrorCode, S3Request, S3Result};
use s3s::{S3Error, S3ErrorCode, S3Request, S3Result, dto::*, s3_error};
use std::collections::HashMap;
#[allow(dead_code)]
@@ -218,11 +218,9 @@ impl S3Access for FS {
let req_info = req.extensions.get_mut::<ReqInfo>().expect("ReqInfo not found");
let (src_bucket, src_key, version_id) = match &req.input.copy_source {
CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
CopySource::Bucket {
ref bucket,
ref key,
version_id,
} => (bucket.to_string(), key.to_string(), version_id.as_ref().map(|v| v.to_string())),
CopySource::Bucket { bucket, key, version_id } => {
(bucket.to_string(), key.to_string(), version_id.as_ref().map(|v| v.to_string()))
}
};
req_info.bucket = Some(src_bucket);
+26 -30
View File
@@ -1,6 +1,6 @@
use common::error::Error;
use ecstore::{disk::error::is_err_file_not_found, error::StorageError};
use s3s::{s3_error, S3Error, S3ErrorCode};
use ecstore::error::StorageError;
use s3s::{S3Error, S3ErrorCode, s3_error};
pub fn to_s3_error(err: Error) -> S3Error {
if let Some(storage_err) = err.downcast_ref::<StorageError>() {
return match storage_err {
@@ -56,18 +56,6 @@ pub fn to_s3_error(err: Error) -> S3Error {
StorageError::ObjectExistsAsDirectory(bucket, object) => {
s3_error!(InvalidArgument, "Object exists on :{} as directory {}", bucket, object)
}
StorageError::InsufficientReadQuorum => {
s3_error!(SlowDown, "Storage resources are insufficient for the read operation")
}
StorageError::InsufficientWriteQuorum => {
s3_error!(SlowDown, "Storage resources are insufficient for the write operation")
}
StorageError::DecommissionNotStarted => s3_error!(InvalidArgument, "Decommission Not Started"),
StorageError::DecommissionAlreadyRunning => s3_error!(InternalError, "Decommission already running"),
StorageError::VolumeNotFound(bucket) => {
s3_error!(NoSuchBucket, "bucket not found {}", bucket)
}
StorageError::InvalidPart(bucket, object, version_id) => {
s3_error!(
InvalidPart,
@@ -187,10 +175,12 @@ mod tests {
let s3_err = to_s3_error(err);
assert_eq!(*s3_err.code(), S3ErrorCode::ServiceUnavailable);
assert!(s3_err
.message()
.unwrap()
.contains("Storage reached its minimum free drive threshold"));
assert!(
s3_err
.message()
.unwrap()
.contains("Storage reached its minimum free drive threshold")
);
}
#[test]
@@ -257,10 +247,12 @@ mod tests {
let s3_err = to_s3_error(err);
assert_eq!(*s3_err.code(), S3ErrorCode::InvalidArgument);
assert!(s3_err
.message()
.unwrap()
.contains("Object name contains forward slash as prefix"));
assert!(
s3_err
.message()
.unwrap()
.contains("Object name contains forward slash as prefix")
);
assert!(s3_err.message().unwrap().contains("test-bucket"));
assert!(s3_err.message().unwrap().contains("/invalid-object"));
}
@@ -358,10 +350,12 @@ mod tests {
let s3_err = to_s3_error(err);
assert_eq!(*s3_err.code(), S3ErrorCode::SlowDown);
assert!(s3_err
.message()
.unwrap()
.contains("Storage resources are insufficient for the read operation"));
assert!(
s3_err
.message()
.unwrap()
.contains("Storage resources are insufficient for the read operation")
);
}
#[test]
@@ -371,10 +365,12 @@ mod tests {
let s3_err = to_s3_error(err);
assert_eq!(*s3_err.code(), S3ErrorCode::SlowDown);
assert!(s3_err
.message()
.unwrap()
.contains("Storage resources are insufficient for the write operation"));
assert!(
s3_err
.message()
.unwrap()
.contains("Storage resources are insufficient for the write operation")
);
}
#[test]