gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 16:48:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

add walk api, fix: load iam config

Browse Source
This commit is contained in:
weisd
2025-01-12 22:36:57 +08:00
parent 121be5eeda
commit bfe8c5e9bf
6 changed files with 606 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/cache_value/metacache_set.rs
+1 -3
View File
@@ -54,7 +54,7 @@ impl Clone for ListPathRawOptions {
}
pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -> Result<()> {
info!("list_path_raw");
// println!("list_path_raw {},{}", &opts.bucket, &opts.path);
if opts.disks.is_empty() {
info!("list_path_raw 0 drives provided");
return Err(Error::from_string("list_path_raw: 0 drives provided"));
@@ -283,11 +283,9 @@ pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -
}
}
info!("read entry should heal: {}", current.name);
if let Some(partial_fn) = opts.partial.as_ref() {
partial_fn(MetaCacheEntries(top_entries), &errs).await;
}
// break;
}
Ok(())
});
ecstore/src/store_list_objects.rs
+536 -219
View File
@@ -1,3 +1,5 @@
use crate::bucket::metadata_sys::get_versioning_config;
use crate::bucket::versioning::VersioningApi;
use crate::cache_value::metacache_set::{list_path_raw, ListPathRawOptions};
use crate::disk::error::{is_all_not_found, is_all_volume_not_found, is_err_eof, DiskError};
use crate::disk::{
@@ -9,10 +11,10 @@ use crate::file_meta::merge_file_meta_versions;
use crate::peer::is_reserved_or_invalid_bucket;
use crate::set_disk::SetDisks;
use crate::store::check_list_objs_args;
use crate::store_api::{ListObjectVersionsInfo, ListObjectsInfo, ObjectInfo, ObjectOptions};
use crate::store_api::{FileInfo, ListObjectVersionsInfo, ListObjectsInfo, ObjectInfo, ObjectOptions};
use crate::store_err::{is_err_bucket_not_found, to_object_err, StorageError};
use crate::utils::path::{self, base_dir_from_prefix, SLASH_SEPARATOR};
use crate::StorageAPI;
use crate::{bucket, StorageAPI};
use crate::{store::ECStore, store_api::ListObjectsV2Info};
use futures::future::join_all;
use rand::seq::SliceRandom;
@@ -22,7 +24,7 @@ use std::io::ErrorKind;
use std::sync::Arc;
use tokio::sync::broadcast::{self, Receiver as B_Receiver};
use tokio::sync::mpsc::{self, Receiver, Sender};
use tracing::error;
use tracing::{error, warn};
use uuid::Uuid;
const MAX_OBJECT_LIST: i32 = 1000;
@@ -245,7 +247,7 @@ impl ECStore {
..Default::default()
};
// warn!("list_objects_generic opts {:?}", &opts);
warn!("list_objects_generic opts {:?}", &opts);
// use get
if !opts.prefix.is_empty() && opts.limit == 1 && opts.marker.is_none() {
@@ -669,6 +671,287 @@ impl ECStore {
Ok(Vec::new())
}
pub async fn walk(
self: Arc<Self>,
rx: B_Receiver<bool>,
bucket: &str,
prefix: &str,
result: Sender<ObjectInfoOrErr>,
opts: WalkOptions,
) -> Result<()> {
check_list_objs_args(bucket, prefix, &None)?;
let mut futures = Vec::new();
let mut inputs = Vec::new();
for eset in self.pools.iter() {
for set in eset.disk_set.iter() {
let (mut disks, infos, _) = set.get_online_disks_with_healing_and_info(true).await;
let rx = rx.resubscribe();
let opts = opts.clone();
let (sender, list_out_rx) = mpsc::channel::<MetaCacheEntry>(1);
inputs.push(list_out_rx);
futures.push(async move {
let mut ask_disks = get_list_quorum(&opts.ask_disks, set.set_drive_count as i32);
if ask_disks == -1 {
let new_disks = get_quorum_disks(&disks, &infos, (disks.len() + 1) / 2);
if !new_disks.is_empty() {
disks = new_disks;
} else {
ask_disks = get_list_quorum("strict", set.set_drive_count as i32);
}
}
if set.set_drive_count == 4 || ask_disks > disks.len() as i32 {
ask_disks = disks.len() as i32;
}
let fallback_disks = {
if ask_disks > 0 && disks.len() > ask_disks as usize {
let mut rand = thread_rng();
disks.shuffle(&mut rand);
disks.split_off(ask_disks as usize)
} else {
Vec::new()
}
};
let listing_quorum = ((ask_disks + 1) / 2) as usize;
let resolver = MetadataResolutionParams {
dir_quorum: listing_quorum,
obj_quorum: listing_quorum,
bucket: bucket.to_owned(),
..Default::default()
};
let path = base_dir_from_prefix(prefix);
let mut filter_prefix = {
prefix
.trim_start_matches(&path)
.trim_start_matches(SLASH_SEPARATOR)
.trim_end_matches(SLASH_SEPARATOR)
.to_owned()
};
if filter_prefix == path {
filter_prefix = "".to_owned();
}
// let (sender, rx1) = mpsc::channel(100);
let tx1 = sender.clone();
let tx2 = sender.clone();
list_path_raw(
rx.resubscribe(),
ListPathRawOptions {
disks: disks.iter().cloned().map(Some).collect(),
fallback_disks: fallback_disks.iter().cloned().map(Some).collect(),
bucket: bucket.to_owned(),
path,
recursice: true,
filter_prefix: Some(filter_prefix),
forward_to: opts.marker.clone(),
min_disks: listing_quorum,
per_disk_limit: opts.limit as i32,
agreed: Some(Box::new(move |entry: MetaCacheEntry| {
Box::pin({
let value = tx1.clone();
async move {
if entry.is_dir() {
return;
}
if let Err(err) = value.send(entry).await {
error!("list_path send fail {:?}", err);
}
}
})
})),
partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<Error>]| {
Box::pin({
let value = tx2.clone();
let resolver = resolver.clone();
async move {
if let Ok(Some(entry)) = entries.resolve(resolver) {
if let Err(err) = value.send(entry).await {
error!("list_path send fail {:?}", err);
}
}
}
})
})),
finished: None,
..Default::default()
},
)
.await
});
}
}
let (merge_tx, mut merge_rx) = mpsc::channel::<MetaCacheEntry>(100);
let bucket = bucket.to_owned();
let vcf = match get_versioning_config(&bucket).await {
Ok((res, _)) => Some(res),
Err(_) => None,
};
tokio::spawn(async move {
let mut sent_err = false;
while let Some(entry) = merge_rx.recv().await {
if opts.latest_only {
let fi = match entry.to_fileinfo(&bucket) {
Ok(res) => res,
Err(err) => {
if !sent_err {
let item = ObjectInfoOrErr {
item: None,
err: Some(err),
};
if let Err(err) = result.send(item).await {
error!("walk result send err {:?}", err);
}
sent_err = true;
return;
}
continue;
}
};
if let Some(fiter) = opts.filter {
if fiter(&fi) {
let item = ObjectInfoOrErr {
item: Some(fi.to_object_info(&bucket, &fi.name, {
if let Some(v) = &vcf {
v.versioned(&fi.name)
} else {
false
}
})),
err: None,
};
if let Err(err) = result.send(item).await {
error!("walk result send err {:?}", err);
}
}
} else {
let item = ObjectInfoOrErr {
item: Some(fi.to_object_info(&bucket, &fi.name, {
if let Some(v) = &vcf {
v.versioned(&fi.name)
} else {
false
}
})),
err: None,
};
if let Err(err) = result.send(item).await {
error!("walk result send err {:?}", err);
}
}
continue;
}
let fvs = match entry.file_info_versions(&bucket) {
Ok(res) => res,
Err(err) => {
let item = ObjectInfoOrErr {
item: None,
err: Some(err),
};
if let Err(err) = result.send(item).await {
error!("walk result send err {:?}", err);
}
return;
}
};
if opts.versions_sort == WalkVersionsSortOrder::Ascending {
//TODO: SORT
}
for fi in fvs.versions.iter() {
if let Some(fiter) = opts.filter {
if fiter(fi) {
let item = ObjectInfoOrErr {
item: Some(fi.to_object_info(&bucket, &fi.name, {
if let Some(v) = &vcf {
v.versioned(&fi.name)
} else {
false
}
})),
err: None,
};
if let Err(err) = result.send(item).await {
error!("walk result send err {:?}", err);
}
}
} else {
let item = ObjectInfoOrErr {
item: Some(fi.to_object_info(&bucket, &fi.name, {
if let Some(v) = &vcf {
v.versioned(&fi.name)
} else {
false
}
})),
err: None,
};
if let Err(err) = result.send(item).await {
error!("walk result send err {:?}", err);
}
}
}
}
});
tokio::spawn(async move { merge_entry_channels(rx, inputs, merge_tx, 1).await });
join_all(futures).await;
Ok(())
}
}
type WalkFilter = fn(&FileInfo) -> bool;
#[derive(Clone, Default)]
pub struct WalkOptions {
pub filter: Option<WalkFilter>, // return WalkFilter returns 'true/false'
pub marker: Option<String>, // set to skip until this object
pub latest_only: bool, // returns only latest versions for all matching objects
pub ask_disks: String, // dictates how many disks are being listed
pub versions_sort: WalkVersionsSortOrder, // sort order for versions of the same object; default: Ascending order in ModTime
pub limit: usize, // maximum number of items, 0 means no limit
}
#[derive(Clone, Default, PartialEq, Eq)]
pub enum WalkVersionsSortOrder {
#[default]
Ascending,
Descending,
}
#[derive(Debug)]
pub struct ObjectInfoOrErr {
pub item: Option<ObjectInfo>,
pub err: Option<Error>,
}
async fn gather_results(
@@ -1092,260 +1375,294 @@ fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 {
// list_path_raw
// #[cfg(test)]
// mod test {
// use std::sync::Arc;
#[cfg(test)]
mod test {
use std::sync::Arc;
// use crate::cache_value::metacache_set::list_path_raw;
// use crate::cache_value::metacache_set::ListPathRawOptions;
// use crate::disk::endpoint::Endpoint;
// use crate::disk::error::is_err_eof;
// use crate::disk::format::FormatV3;
// use crate::disk::new_disk;
// use crate::disk::DiskAPI;
// use crate::disk::DiskOption;
// use crate::disk::MetaCacheEntries;
// use crate::disk::MetaCacheEntry;
// use crate::disk::WalkDirOptions;
// use crate::endpoints::EndpointServerPools;
// use crate::error::Error;
// use crate::metacache::writer::MetacacheReader;
// use crate::set_disk::SetDisks;
// use crate::store::ECStore;
// use crate::store_list_objects::ListPathOptions;
// use futures::future::join_all;
// use lock::namespace_lock::NsLockMap;
// use tokio::sync::broadcast;
// use tokio::sync::mpsc;
// use tokio::sync::RwLock;
// use uuid::Uuid;
use crate::cache_value::metacache_set::list_path_raw;
use crate::cache_value::metacache_set::ListPathRawOptions;
use crate::disk::endpoint::Endpoint;
use crate::disk::error::is_err_eof;
use crate::disk::format::FormatV3;
use crate::disk::new_disk;
use crate::disk::DiskAPI;
use crate::disk::DiskOption;
use crate::disk::MetaCacheEntries;
use crate::disk::MetaCacheEntry;
use crate::disk::WalkDirOptions;
use crate::endpoints::EndpointServerPools;
use crate::error::Error;
use crate::metacache::writer::MetacacheReader;
use crate::set_disk::SetDisks;
use crate::store::ECStore;
use crate::store_list_objects::ListPathOptions;
use crate::store_list_objects::WalkOptions;
use crate::store_list_objects::WalkVersionsSortOrder;
use futures::future::join_all;
use lock::namespace_lock::NsLockMap;
use tokio::sync::broadcast;
use tokio::sync::mpsc;
use tokio::sync::RwLock;
use uuid::Uuid;
// #[tokio::test]
// async fn test_walk_dir() {
// let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
// ep.pool_idx = 0;
// ep.set_idx = 0;
// ep.disk_idx = 0;
// ep.is_local = true;
// #[tokio::test]
// async fn test_walk_dir() {
// let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
// ep.pool_idx = 0;
// ep.set_idx = 0;
// ep.disk_idx = 0;
// ep.is_local = true;
// let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail");
// let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail");
// // let disk = match LocalDisk::new(&ep, false).await {
// // Ok(res) => res,
// // Err(err) => {
// // println!("LocalDisk::new err {:?}", err);
// // return;
// // }
// // };
// // let disk = match LocalDisk::new(&ep, false).await {
// // Ok(res) => res,
// // Err(err) => {
// // println!("LocalDisk::new err {:?}", err);
// // return;
// // }
// // };
// let (rd, mut wr) = tokio::io::duplex(64);
// let (rd, mut wr) = tokio::io::duplex(64);
// let job = tokio::spawn(async move {
// let opts = WalkDirOptions {
// bucket: "dada".to_owned(),
// base_dir: "".to_owned(),
// recursive: true,
// ..Default::default()
// };
// let job = tokio::spawn(async move {
// let opts = WalkDirOptions {
// bucket: "dada".to_owned(),
// base_dir: "".to_owned(),
// recursive: true,
// ..Default::default()
// };
// println!("walk opts {:?}", opts);
// if let Err(err) = disk.walk_dir(opts, &mut wr).await {
// println!("walk_dir err {:?}", err);
// }
// });
// println!("walk opts {:?}", opts);
// if let Err(err) = disk.walk_dir(opts, &mut wr).await {
// println!("walk_dir err {:?}", err);
// }
// });
// let job2 = tokio::spawn(async move {
// let mut mrd = MetacacheReader::new(rd);
// let job2 = tokio::spawn(async move {
// let mut mrd = MetacacheReader::new(rd);
// loop {
// match mrd.peek().await {
// Ok(res) => {
// if let Some(info) = res {
// println!("info {:?}", info.name)
// } else {
// break;
// }
// }
// Err(err) => {
// if is_err_eof(&err) {
// break;
// }
// loop {
// match mrd.peek().await {
// Ok(res) => {
// if let Some(info) = res {
// println!("info {:?}", info.name)
// } else {
// break;
// }
// }
// Err(err) => {
// if is_err_eof(&err) {
// break;
// }
// println!("get err {:?}", err);
// break;
// }
// }
// }
// });
// join_all(vec![job, job2]).await;
// }
// println!("get err {:?}", err);
// break;
// }
// }
// }
// });
// join_all(vec![job, job2]).await;
// }
// #[tokio::test]
// async fn test_list_path_raw() {
// let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
// ep.pool_idx = 0;
// ep.set_idx = 0;
// ep.disk_idx = 0;
// ep.is_local = true;
// #[tokio::test]
// async fn test_list_path_raw() {
// let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
// ep.pool_idx = 0;
// ep.set_idx = 0;
// ep.disk_idx = 0;
// ep.is_local = true;
// let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail");
// let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail");
// // let disk = match LocalDisk::new(&ep, false).await {
// // Ok(res) => res,
// // Err(err) => {
// // println!("LocalDisk::new err {:?}", err);
// // return;
// // }
// // };
// // let disk = match LocalDisk::new(&ep, false).await {
// // Ok(res) => res,
// // Err(err) => {
// // println!("LocalDisk::new err {:?}", err);
// // return;
// // }
// // };
// let (_, rx) = broadcast::channel(1);
// let bucket = "dada".to_owned();
// let forward_to = None;
// let disks = vec![Some(disk)];
// let fallback_disks = Vec::new();
// let (_, rx) = broadcast::channel(1);
// let bucket = "dada".to_owned();
// let forward_to = None;
// let disks = vec![Some(disk)];
// let fallback_disks = Vec::new();
// list_path_raw(
// rx,
// ListPathRawOptions {
// disks,
// fallback_disks,
// bucket,
// path: "".to_owned(),
// recursice: true,
// forward_to,
// min_disks: 1,
// report_not_found: false,
// agreed: Some(Box::new(move |entry: MetaCacheEntry| {
// Box::pin(async move { println!("get entry: {}", entry.name) })
// })),
// partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<Error>]| {
// Box::pin(async move { println!("get entries: {:?}", entries) })
// })),
// finished: None,
// ..Default::default()
// },
// )
// .await
// .unwrap();
// }
// list_path_raw(
// rx,
// ListPathRawOptions {
// disks,
// fallback_disks,
// bucket,
// path: "".to_owned(),
// recursice: true,
// forward_to,
// min_disks: 1,
// report_not_found: false,
// agreed: Some(Box::new(move |entry: MetaCacheEntry| {
// Box::pin(async move { println!("get entry: {}", entry.name) })
// })),
// partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option<Error>]| {
// Box::pin(async move { println!("get entries: {:?}", entries) })
// })),
// finished: None,
// ..Default::default()
// },
// )
// .await
// .unwrap();
// }
// #[tokio::test]
// async fn test_set_list_path() {
// let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
// ep.pool_idx = 0;
// ep.set_idx = 0;
// ep.disk_idx = 0;
// ep.is_local = true;
// #[tokio::test]
// async fn test_set_list_path() {
// let mut ep = Endpoint::try_from("/Users/weisd/project/weisd/s3-rustfs/target/volume/test").unwrap();
// ep.pool_idx = 0;
// ep.set_idx = 0;
// ep.disk_idx = 0;
// ep.is_local = true;
// let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail");
// let _ = disk.set_disk_id(Some(Uuid::new_v4())).await;
// let disk = new_disk(&ep, &DiskOption::default()).await.expect("init disk fail");
// let _ = disk.set_disk_id(Some(Uuid::new_v4())).await;
// let set = SetDisks {
// lockers: Vec::new(),
// locker_owner: String::new(),
// ns_mutex: Arc::new(RwLock::new(NsLockMap::new(false))),
// disks: RwLock::new(vec![Some(disk)]),
// set_endpoints: Vec::new(),
// set_drive_count: 1,
// default_parity_count: 0,
// set_index: 0,
// pool_index: 0,
// format: FormatV3::new(1, 1),
// };
// let set = SetDisks {
// lockers: Vec::new(),
// locker_owner: String::new(),
// ns_mutex: Arc::new(RwLock::new(NsLockMap::new(false))),
// disks: RwLock::new(vec![Some(disk)]),
// set_endpoints: Vec::new(),
// set_drive_count: 1,
// default_parity_count: 0,
// set_index: 0,
// pool_index: 0,
// format: FormatV3::new(1, 1),
// };
// let (_tx, rx) = broadcast::channel(1);
// let (_tx, rx) = broadcast::channel(1);
// let bucket = "dada".to_owned();
// let bucket = "dada".to_owned();
// let opts = ListPathOptions {
// bucket,
// recursive: true,
// ..Default::default()
// };
// let opts = ListPathOptions {
// bucket,
// recursive: true,
// ..Default::default()
// };
// let (sender, mut recv) = mpsc::channel(10);
// let (sender, mut recv) = mpsc::channel(10);
// set.list_path(rx, opts, sender).await.unwrap();
// set.list_path(rx, opts, sender).await.unwrap();
// while let Some(entry) = recv.recv().await {
// println!("get entry {:?}", entry.name)
// }
// }
// while let Some(entry) = recv.recv().await {
// println!("get entry {:?}", entry.name)
// }
// }
// #[tokio::test]
// async fn test_list_merged() {
// let server_address = "localhost:9000";
// #[tokio::test]
//walk() {
// let server_address = "localhost:9000";
// let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
// server_address,
// vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
// )
// .unwrap();
// let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
// server_address,
// vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
// )
// .unwrap();
// let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
// .await
// .unwrap();
// let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
// .await
// .unwrap();
// let (_tx, rx) = broadcast::channel(1);
// let (_tx, rx) = broadcast::channel(1);
// let bucket = "dada".to_owned();
// let opts = ListPathOptions {
// bucket,
// recursive: true,
// ..Default::default()
// };
// let bucket = "dada".to_owned();
// let opts = ListPathOptions {
// bucket,
// recursive: true,
// ..Default::default()
// };
// let (sender, mut recv) = mpsc::channel(10);
// let (sender, mut recv) = mpsc::channel(10);
// store.list_merged(rx, opts, sender).await.unwrap();
// store.list_merged(rx, opts, sender).await.unwrap();
// while let Some(entry) = recv.recv().await {
// println!("get entry {:?}", entry.name)
// }
// }
// while let Some(entry) = recv.recv().await {
// println!("get entry {:?}", entry.name)
// }
// }
// #[tokio::test]
// async fn test_list_path() {
// let server_address = "localhost:9000";
// #[tokio::test]
// async fn test_list_path() {
// let server_address = "localhost:9000";
// let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
// server_address,
// vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
// )
// .unwrap();
// let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
// server_address,
// vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
// )
// .unwrap();
// let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
// .await
// .unwrap();
// let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
// .await
// .unwrap();
// let bucket = "dada".to_owned();
// let opts = ListPathOptions {
// bucket,
// recursive: true,
// limit: 100,
// let bucket = "dada".to_owned();
// let opts = ListPathOptions {
// bucket,
// recursive: true,
// limit: 100,
// ..Default::default()
// };
// ..Default::default()
// };
// let ret = store.list_path(&opts).await.unwrap();
// println!("ret {:?}", ret);
// }
// let ret = store.list_path(&opts).await.unwrap();
// println!("ret {:?}", ret);
// }
// // #[tokio::test]
// // async fn test_list_objects_v2() {
// // let server_address = "localhost:9000";
// #[tokio::test]
// async fn test_list_objects_v2() {
// let server_address = "localhost:9000";
// // let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
// // server_address,
// // vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
// // )
// // .unwrap();
// let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
// server_address,
// vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
// )
// .unwrap();
// // let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
// // .await
// // .unwrap();
// let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
// .await
// .unwrap();
// // let ret = store.list_objects_v2("data", "", "", "", 100, false, "").await.unwrap();
// // println!("ret {:?}", ret);
// // }
// }
// let ret = store.list_objects_v2("data", "", "", "", 100, false, "").await.unwrap();
// println!("ret {:?}", ret);
// }
#[tokio::test]
async fn test_walk() {
let server_address = "localhost:9000";
let (endpoint_pools, _setup_type) = EndpointServerPools::from_volumes(
server_address,
vec!["/Users/weisd/project/weisd/s3-rustfs/target/volume/test".to_string()],
)
.unwrap();
let store = ECStore::new(server_address.to_string(), endpoint_pools.clone())
.await
.unwrap();
ECStore::init(store.clone()).await.unwrap();
let (_tx, rx) = broadcast::channel(1);
let bucket = ".rustfs.sys";
let prefix = "config/iam/sts/";
let (sender, mut recv) = mpsc::channel(10);
let opts = WalkOptions::default();
store.walk(rx, bucket, prefix, sender, opts).await.unwrap();
while let Some(entry) = recv.recv().await {
println!("get entry {:?}", entry)
}
}
}
iam/src/manager.rs
+1 -1
View File
@@ -261,7 +261,7 @@ where
self.api.save_iam_config(&user_entiry, path).await?;
Cache::add_or_update(
&self.cache.users,
&self.cache.sts_accounts,
&user_entiry.credentials.access_key,
&user_entiry,
OffsetDateTime::now_utc(),
iam/src/store/object.rs
+61 -51
View File
@@ -4,12 +4,14 @@ use ecstore::{
config::error::is_err_config_not_found,
store::ECStore,
store_api::{ObjectIO, ObjectInfo, ObjectOptions, PutObjReader},
utils::path::dir,
StorageAPI,
store_list_objects::{ObjectInfoOrErr, WalkOptions},
utils::path::{dir, SLASH_SEPARATOR},
};
use futures::future::try_join_all;
use log::{debug, warn};
use serde::{de::DeserializeOwned, Serialize};
use tokio::sync::broadcast;
use tokio::sync::mpsc::{self};
use super::Store;
use crate::{
@@ -31,45 +33,54 @@ impl ObjectStore {
Self { object_api }
}
async fn list_iam_config_items(&self, prefix: &str, items: &[&str]) -> crate::Result<Vec<String>> {
debug!("list iam config items, prefix: {prefix}");
async fn list_iam_config_items(&self, prefix: &str) -> crate::Result<Vec<String>> {
debug!("list iam config items, prefix: {}", &prefix);
// todo, 实现walk，使用walk
let mut futures = Vec::with_capacity(items.len());
for item in items {
let prefix = format!("{}{}", prefix, item);
futures.push(async move {
// let items = self
// .object_api
// .clone()
// .list_path(&ListPathOptions {
// bucket: Self::BUCKET_NAME.into(),
// prefix: prefix.clone(),
// ..Default::default()
// })
// .await;
let (ctx_tx, ctx_rx) = broadcast::channel(1);
let items = self
.object_api
.clone()
.list_objects_v2(Self::BUCKET_NAME, &prefix.clone(), None, None, 0, false, None)
.await;
// let prefix = format!("{}{}", prefix, item);
match items {
Ok(items) => Result::<_, crate::Error>::Ok(items.prefixes),
Err(e) => {
if is_err_config_not_found(&e) {
Result::<_, crate::Error>::Ok(vec![])
} else {
Err(Error::StringError(format!("list {prefix} failed, err: {e:?}")))
}
}
}
});
let ctx_rx = ctx_rx.resubscribe();
let (tx, mut rx) = mpsc::channel::<ObjectInfoOrErr>(100);
let store = self.object_api.clone();
let path = prefix.to_owned();
tokio::spawn(async move { store.walk(ctx_rx, Self::BUCKET_NAME, &path, tx, WalkOptions::default()).await });
let mut ret = Vec::new();
while let Some(v) = rx.recv().await {
if let Some(err) = v.err {
warn!("list_iam_config_items {:?}", err);
let _ = ctx_tx.send(true);
return Err(Error::EcstoreError(err));
}
if let Some(item) = v.item {
let name = item.name.trim_start_matches(prefix).trim_end_matches(SLASH_SEPARATOR);
ret.push(name.to_owned());
}
}
Ok(ret)
// match items {
// Ok(items) => Result::<_, crate::Error>::Ok(items.prefixes),
// Err(e) => {
// if is_err_config_not_found(&e) {
// Result::<_, crate::Error>::Ok(vec![])
// } else {
// Err(Error::StringError(format!("list {prefix} failed, err: {e:?}")))
// }
// }
// }
// TODO: FIXME:
Ok(try_join_all(futures).await?.into_iter().flat_map(|x| x.into_iter()).collect())
// Ok(try_join_all(futures).await?.into_iter().flat_map(|x| x.into_iter()).collect())
}
async fn load_policy(&self, name: &str) -> crate::Result<PolicyDoc> {
@@ -152,7 +163,7 @@ impl Store for ObjectStore {
}
async fn load_policy_docs(&self) -> crate::Result<HashMap<String, PolicyDoc>> {
let paths = self.list_iam_config_items("config/iam/", &["policies/"]).await?;
let paths = self.list_iam_config_items("config/iam/policies/").await?;
let mut result = Self::get_default_policyes();
for path in paths {
@@ -174,7 +185,9 @@ impl Store for ObjectStore {
}
async fn load_users(&self, user_type: UserType) -> crate::Result<HashMap<String, UserIdentity>> {
let paths = self.list_iam_config_items("config/iam/", &[user_type.prefix()]).await?;
let paths = self
.list_iam_config_items(format!("{}{}", "config/iam/", user_type.prefix()).as_str())
.await?;
let mut result = HashMap::new();
for path in paths {
@@ -202,21 +215,18 @@ impl Store for ObjectStore {
/// load all and make a new cache.
async fn load_all(&self, cache: &Cache) -> crate::Result<()> {
let items = self
.list_iam_config_items(
"config/iam/",
&[
"policydb/",
"policies/",
"groups/",
"policydb/users/",
"policydb/groups/",
"service-accounts/",
"policydb/sts-users/",
"sts/",
],
)
.await?;
let _items = &[
"policydb/",
"policies/",
"groups/",
"policydb/users/",
"policydb/groups/",
"service-accounts/",
"policydb/sts-users/",
"sts/",
];
let items = self.list_iam_config_items("config/iam/").await?;
debug!("all iam items: {items:?}");
let (policy_docs, users, user_policies, sts_policies, sts_accounts) = (
rustfs/src/admin/handlers/user.rs
-4
View File
@@ -66,10 +66,6 @@ impl Operation for AddUser {
}
}
fn check_claims_from_token(_token: &str) -> bool {
true
}
pub struct SetUserStatus {}
#[async_trait::async_trait]
impl Operation for SetUserStatus {
rustfs/src/auth.rs
+7
View File
@@ -1,4 +1,5 @@
use iam::cache::CacheInner;
use log::warn;
use s3s::auth::S3Auth;
use s3s::auth::SecretKey;
use s3s::auth::SimpleAuth;
@@ -27,9 +28,15 @@ impl S3Auth for IAMAuth {
return Ok(key);
}
warn!("Failed to get secret key from simple auth");
if let Ok(iam_store) = iam::get() {
let c = CacheInner::from(&iam_store.cache);
warn!("Failed to get secret key from simple auth, try cache {}", access_key);
warn!("users {:?}", c.users.values());
warn!("sts_accounts {:?}", c.sts_accounts.values());
if let Some(id) = c.get_user(access_key) {
warn!("get cred {:?}", id.credentials);
return Ok(SecretKey::from(id.credentials.secret_key.clone()));
}
}