diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index a392efe8f..2222f1c24 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -12,15 +12,13 @@ use crate::{ }; use bytes::Bytes; use path_absolutize::Absolutize; -use std::sync::Arc; use std::{ fs::Metadata, path::{Path, PathBuf}, }; use time::OffsetDateTime; use tokio::fs::{self, File}; -use tokio::io::{DuplexStream, ErrorKind}; -use tokio::sync::mpsc; +use tokio::io::ErrorKind; use tracing::{debug, warn}; use uuid::Uuid; @@ -526,7 +524,8 @@ impl DiskAPI for LocalDisk { Ok(volumes) } - async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc) -> Result<()> { + + async fn walk_dir(&self, opts: WalkDirOptions) -> Result> { let mut entries = self.list_dir("", &opts.bucket, &opts.base_dir, -1).await?; entries.sort(); @@ -536,11 +535,13 @@ impl DiskAPI for LocalDisk { let bucket = opts.bucket.as_str(); + let mut metas = Vec::new(); + // 第一层过滤 for entry in entries.iter() { // check limit if opts.limit > 0 && objs_returned >= opts.limit { - return Ok(()); + return Ok(metas); } // check prefix if !opts.filter_prefix.is_empty() && !entry.starts_with(&opts.filter_prefix) { @@ -558,7 +559,7 @@ impl DiskAPI for LocalDisk { let (fdata, _) = match self.read_metadata_with_dmtime(&fpath).await { Ok(res) => res, - Err(e) => { + Err(_) => { // TODO: check err (Vec::new(), OffsetDateTime::UNIX_EPOCH) } @@ -566,10 +567,10 @@ impl DiskAPI for LocalDisk { meta.metadata = fdata; - // TODO: FIXME: + metas.push(meta); } - Ok(()) + Ok(metas) } // #[tracing::instrument(skip(self))] diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 7489cdfe7..b694f99b8 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -22,7 +22,7 @@ use std::{fmt::Debug, io::SeekFrom, pin::Pin, sync::Arc}; use time::OffsetDateTime; use tokio::{ fs::File, - io::{AsyncReadExt, AsyncSeekExt, AsyncWrite, DuplexStream}, + io::{AsyncReadExt, AsyncSeekExt, AsyncWrite}, }; use uuid::Uuid; @@ -53,7 +53,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { // 读目录下的所有文件、目录 async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result>; // 并发边读边写 TODO: wr io.Writer - async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc) -> Result<()>; + async fn walk_dir(&self, opts: WalkDirOptions) -> Result>; async fn rename_data( &self, src_volume: &str, @@ -112,16 +112,71 @@ pub struct WalkDirOptions { #[derive(Debug, Default)] pub struct MetaCacheEntry { // name is the full name of the object including prefixes - name: String, + pub name: String, // Metadata. If none is present it is not an object but only a prefix. // Entries without metadata will only be present in non-recursive scans. - metadata: Vec, + pub metadata: Vec, // cached contains the metadata if decoded. cached: Option, // Indicates the entry can be reused and only one reference to metadata is expected. - reusable: bool, + _reusable: bool, +} + +impl MetaCacheEntry { + pub fn marshal_msg(&self) -> Result> { + let mut wr = Vec::new(); + rmp::encode::write_bool(&mut wr, true)?; + + rmp::encode::write_str(&mut wr, &self.name)?; + + rmp::encode::write_bin(&mut wr, &self.metadata)?; + + Ok(wr) + } + pub fn is_dir(&self) -> bool { + self.metadata.is_empty() && self.name.ends_with("/") + } + pub fn is_object(&self) -> bool { + !self.metadata.is_empty() + } + + #[tracing::instrument(level = "debug", skip(self))] + pub fn to_fileinfo(&self, bucket: &str) -> Result> { + if self.is_dir() { + return Ok(Some(FileInfo { + volume: bucket.to_owned(), + name: self.name.clone(), + ..Default::default() + })); + } + + if self.cached.is_some() { + let fm = self.cached.as_ref().unwrap(); + if fm.versions.is_empty() { + return Ok(Some(FileInfo { + volume: bucket.to_owned(), + name: self.name.clone(), + deleted: true, + is_latest: true, + mod_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + })); + } + + let fi = fm.into_fileinfo(bucket, self.name.as_str(), "", false, false)?; + + return Ok(Some(fi)); + } + + let mut fm = FileMeta::new(); + fm.unmarshal_msg(&self.metadata)?; + + let fi = fm.into_fileinfo(bucket, self.name.as_str(), "", false, false)?; + + return Ok(Some(fi)); + } } pub struct DiskOption { diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 25e2735d6..b9adb1f9d 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -10,8 +10,8 @@ use crate::{ error::Result, set_disk::SetDisks, store_api::{ - BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsInfo, ListObjectsV2Info, - MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI, + BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsV2Info, MakeBucketOptions, + MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI, }, utils::hash, }; @@ -137,13 +137,13 @@ impl StorageAPI for Sets { async fn list_objects_v2( &self, - bucket: &str, - prefix: &str, - continuation_token: &str, - delimiter: &str, - max_keys: i32, - fetch_owner: bool, - start_after: &str, + _bucket: &str, + _prefix: &str, + _continuation_token: &str, + _delimiter: &str, + _max_keys: i32, + _fetch_owner: bool, + _start_after: &str, ) -> Result { unimplemented!() } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 5f12b15e5..2f990c133 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -1,6 +1,6 @@ use crate::{ bucket_meta::BucketMetadata, - disk::{error::DiskError, DiskOption, DiskStore, MetaCacheEntry, WalkDirOptions, RUSTFS_META_BUCKET}, + disk::{error::DiskError, DiskOption, DiskStore, WalkDirOptions, RUSTFS_META_BUCKET}, disks_layout::DisksLayout, endpoints::EndpointServerPools, error::{Error, Result}, @@ -15,8 +15,8 @@ use crate::{ use futures::future::join_all; use http::HeaderMap; use s3s::{dto::StreamingBlob, Body}; -use std::{collections::HashMap, sync::Arc}; -use tokio::sync::mpsc; +use std::collections::{HashMap, HashSet}; +use tracing::warn; use uuid::Uuid; #[derive(Debug)] @@ -108,22 +108,24 @@ impl ECStore { self.pools.len() == 1 } - async fn list_path(&self, opts: &ListPathOptions) -> Result<()> { - let (sender, rec) = mpsc::channel(64); - self.listMerged(opts, sender).await?; - Ok(()) + async fn list_path(&self, opts: &ListPathOptions) -> Result { + let objects = self.list_merged(opts).await?; + + let info = ListObjectsInfo { + objects, + ..Default::default() + }; + Ok(info) } // 读所有 - async fn listMerged(&self, opts: &ListPathOptions, results: mpsc::Sender) -> Result<()> { + async fn list_merged(&self, opts: &ListPathOptions) -> Result> { let opts = WalkDirOptions { bucket: opts.bucket.clone(), ..Default::default() }; - let (mut wr, mut rd) = tokio::io::duplex(1024); - - let wr = Arc::new(wr); + // let (mut wr, mut rd) = tokio::io::duplex(1024); let mut futures = Vec::new(); @@ -136,8 +138,8 @@ impl ECStore { let disk = disk.as_ref().unwrap(); let opts = opts.clone(); - let wr = wr.clone(); - futures.push(disk.walk_dir(opts, wr)); + // let mut wr = &mut wr; + futures.push(disk.walk_dir(opts)); // tokio::spawn(async move { disk.walk_dir(opts, wr).await }); } } @@ -146,20 +148,52 @@ impl ECStore { let results = join_all(futures).await; let mut errs = Vec::new(); + let mut ress = Vec::new(); + let mut uniq = HashSet::new(); for res in results { match res { - Ok(_) => errs.push(None), + Ok(entrys) => { + for entry in entrys { + if !uniq.contains(&entry.name) { + uniq.insert(entry.name.clone()); + // TODO: 过滤 + if opts.limit > 0 && ress.len() as i32 >= opts.limit { + return Ok(ress); + } + + if entry.is_object() { + let fi = entry.to_fileinfo(&opts.bucket)?; + if fi.is_some() { + ress.push(fi.unwrap().into_object_info(&opts.bucket, &entry.name, false)); + } + continue; + } + + if entry.is_dir() { + ress.push(ObjectInfo { + is_dir: true, + bucket: opts.bucket.clone(), + name: entry.name, + ..Default::default() + }); + } + } + } + errs.push(None); + } Err(e) => errs.push(Some(e)), } } - Ok(()) + warn!("list_merged errs {:?}", errs); + + Ok(ress) } } #[derive(Debug, Default)] -struct ListPathOptions { +pub struct ListPathOptions { pub id: String, // Bucket of the listing. @@ -234,12 +268,12 @@ impl StorageAPI for ECStore { async fn list_objects_v2( &self, bucket: &str, - prefix: &str, + _prefix: &str, continuation_token: &str, - delimiter: &str, + _delimiter: &str, max_keys: i32, - fetch_owner: bool, - start_after: &str, + _fetch_owner: bool, + _start_after: &str, ) -> Result { let opts = ListPathOptions { bucket: bucket.to_string(), @@ -247,9 +281,19 @@ impl StorageAPI for ECStore { ..Default::default() }; - self.list_path(&opts).await?; + let info = self.list_path(&opts).await?; - Ok(ListObjectsV2Info::default()) + warn!("list_objects_v2 info {:?}", info); + + let v2 = ListObjectsV2Info { + is_truncated: info.is_truncated, + continuation_token: continuation_token.to_owned(), + next_continuation_token: info.next_marker, + objects: info.objects, + prefixes: info.prefixes, + }; + + Ok(v2) } async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { let object = utils::path::encode_dir_object(object); diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 1b68aae73..3aa50f569 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -409,7 +409,7 @@ impl From for CompletePart { } } -#[derive(Debug)] +#[derive(Debug, Default)] pub struct ObjectInfo { pub bucket: String, pub name: String, @@ -424,7 +424,7 @@ pub struct ObjectInfo { pub is_latest: bool, } -#[derive(Debug)] +#[derive(Debug, Default)] pub struct ListObjectsInfo { // Indicates whether the returned list objects response is truncated. A // value of true indicates that the list was truncated. The list can be truncated diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index c75bd1904..3f6fd6d3d 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -10,7 +10,6 @@ use ecstore::store_api::PutObjReader; use ecstore::store_api::StorageAPI; use futures::pin_mut; use futures::{Stream, StreamExt}; -use futures_util::TryStreamExt; use http::HeaderMap; use s3s::dto::*; use s3s::s3_error; @@ -20,9 +19,8 @@ use s3s::S3Result; use s3s::S3; use s3s::{S3Request, S3Response}; use std::fmt::Debug; -use std::pin::Pin; use std::str::FromStr; -use std::task::Poll; +use tracing::warn; use transform_stream::AsyncTryStream; use ecstore::error::Result; @@ -258,13 +256,16 @@ impl S3 for FS { .. } = req.input; - let _object_infos = try_!( + let prefix = prefix.unwrap_or_default(); + let delimiter = delimiter.unwrap_or_default(); + + let object_infos = try_!( self.store .list_objects_v2( &bucket, - &prefix.unwrap_or_default(), + &prefix, &continuation_token.unwrap_or_default(), - &delimiter.unwrap_or_default(), + &delimiter, max_keys.unwrap_or_default(), fetch_owner.unwrap_or_default(), &start_after.unwrap_or_default() @@ -272,7 +273,41 @@ impl S3 for FS { .await ); - let output = ListObjectsV2Output { ..Default::default() }; + warn!("object_infos {:?}", object_infos); + + let objects: Vec = object_infos + .objects + .iter() + .filter(|v| !v.name.is_empty()) + .map(|v| { + let mut obj = Object::default(); + obj.key = Some(v.name.to_owned()); + obj.last_modified = v.mod_time.map(Timestamp::from); + obj.size = Some(v.size as i64); + + if fetch_owner.is_some_and(|v| v) { + obj.owner = Some(Owner { + display_name: Some("rustfs".to_owned()), + id: Some("v0.1".to_owned()), + }); + } + obj + }) + .collect(); + + let key_count = objects.len() as i32; + + let output = ListObjectsV2Output { + key_count: Some(key_count), + max_keys: Some(key_count), + contents: Some(objects), + delimiter: Some(delimiter), + name: Some(bucket), + prefix: Some(prefix), + ..Default::default() + }; + + // let output = ListObjectsV2Output { ..Default::default() }; Ok(S3Response::new(output)) }