init listobject

TODO.md

@@ -2,12 +2,12 @@
 
 ## 基础存储
 
-- [ ] 优化xlmeta, 自定义msg数据结构
+- [x] 优化xlmeta, 自定义msg数据结构
 - [ ] 小文件存储到metafile, inlinedata
-- [ ] 上传同名文件时，删除旧版本文件
+- [x] 上传同名文件时，删除旧版本文件
 - [ ] EC可用读写数量判断 Read/WriteQuorum
 - [ ] 错误类型判断,程序中判断错误类型，如何统一错误
-- [ ] 优化并发执行
+- [ ] 优化并发执行，边读边取，可中断
 - [ ] 抽象出metafile存储
 - [ ] 代码优化
 

ecstore/src/disk/local.rs

@@ -1,7 +1,9 @@
 use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
 use super::{
-    DeleteOptions, DiskAPI, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, VolumeInfo,
+    DeleteOptions, DiskAPI, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions,
+    RenameDataResp, VolumeInfo, WalkDirOptions,
 };
+use crate::disk::STORAGE_FORMAT_FILE;
 use crate::{
     error::{Error, Result},
     file_meta::FileMeta,
@@ -10,13 +12,15 @@ 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::ErrorKind;
+use tokio::io::{DuplexStream, ErrorKind};
+use tokio::sync::mpsc;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
@@ -496,7 +500,7 @@ impl DiskAPI for LocalDisk {
 
         // Ok((buffer, bytes_read))
     }
-    async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: usize) -> Result<Vec<String>> {
+    async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result<Vec<String>> {
         let p = self.get_bucket_path(volume)?;
 
         let mut entries = fs::read_dir(&p).await?;
@@ -522,8 +526,50 @@ impl DiskAPI for LocalDisk {
 
         Ok(volumes)
     }
-    async fn walk_dir(&self) -> Result<Vec<FileInfo>> {
-        unimplemented!()
+    async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc<DuplexStream>) -> Result<()> {
+        let mut entries = self.list_dir("", &opts.bucket, &opts.base_dir, -1).await?;
+
+        entries.sort();
+
+        // 已读计数
+        let objs_returned = 0;
+
+        let bucket = opts.bucket.as_str();
+
+        // 第一层过滤
+        for entry in entries.iter() {
+            // check limit
+            if opts.limit > 0 && objs_returned >= opts.limit {
+                return Ok(());
+            }
+            // check prefix
+            if !opts.filter_prefix.is_empty() && !entry.starts_with(&opts.filter_prefix) {
+                continue;
+            }
+
+            warn!("walk_dir entry {}", entry);
+
+            let mut meta = MetaCacheEntry {
+                name: entry.clone(),
+                ..Default::default()
+            };
+
+            let fpath = self.get_object_path(bucket, format!("{}/{}", &meta.name, STORAGE_FORMAT_FILE).as_str())?;
+
+            let (fdata, _) = match self.read_metadata_with_dmtime(&fpath).await {
+                Ok(res) => res,
+                Err(e) => {
+                    // TODO: check err
+                    (Vec::new(), OffsetDateTime::UNIX_EPOCH)
+                }
+            };
+
+            meta.metadata = fdata;
+
+            // TODO: FIXME:
+        }
+
+        Ok(())
     }
 
     // #[tracing::instrument(skip(self))]

ecstore/src/disk/mod.rs

@@ -14,6 +14,7 @@ const STORAGE_FORMAT_FILE: &str = "xl.meta";
 use crate::{
     erasure::ReadAt,
     error::Result,
+    file_meta::FileMeta,
     store_api::{FileInfo, RawFileInfo},
 };
 use bytes::Bytes;
@@ -21,7 +22,7 @@ use std::{fmt::Debug, io::SeekFrom, pin::Pin, sync::Arc};
 use time::OffsetDateTime;
 use tokio::{
     fs::File,
-    io::{AsyncReadExt, AsyncSeekExt, AsyncWrite},
+    io::{AsyncReadExt, AsyncSeekExt, AsyncWrite, DuplexStream},
 };
 use uuid::Uuid;
 
@@ -50,9 +51,9 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
     async fn read_file(&self, volume: &str, path: &str) -> Result<FileReader>;
     // 读目录下的所有文件、目录
-    async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: usize) -> Result<Vec<String>>;
-    // 读目录下的所有xl.meta
-    async fn walk_dir(&self) -> Result<Vec<FileInfo>>;
+    async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result<Vec<String>>;
+    // 并发边读边写 TODO: wr io.Writer
+    async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc<DuplexStream>) -> Result<()>;
     async fn rename_data(
         &self,
         src_volume: &str,
@@ -81,6 +82,48 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
 }
 
+#[derive(Debug, Default, Clone)]
+pub struct WalkDirOptions {
+    // Bucket to scanner
+    pub bucket: String,
+    // Directory inside the bucket.
+    pub base_dir: String,
+    // Do a full recursive scan.
+    pub recursive: bool,
+
+    // ReportNotFound will return errFileNotFound if all disks reports the BaseDir cannot be found.
+    pub report_notfound: bool,
+
+    // FilterPrefix will only return results with given prefix within folder.
+    // Should never contain a slash.
+    pub filter_prefix: String,
+
+    // ForwardTo will forward to the given object path.
+    pub forward_to: String,
+
+    // Limit the number of returned objects if > 0.
+    pub limit: i32,
+
+    // DiskID contains the disk ID of the disk.
+    // Leave empty to not check disk ID.
+    pub disk_id: String,
+}
+
+#[derive(Debug, Default)]
+pub struct MetaCacheEntry {
+    // name is the full name of the object including prefixes
+    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<u8>,
+
+    // cached contains the metadata if decoded.
+    cached: Option<FileMeta>,
+
+    // Indicates the entry can be reused and only one reference to metadata is expected.
+    reusable: bool,
+}
+
 pub struct DiskOption {
     pub cleanup: bool,
     pub health_check: bool,

ecstore/src/sets.rs

@@ -2,14 +2,16 @@ use http::HeaderMap;
 use uuid::Uuid;
 
 use crate::{
-    disk::format::{DistributionAlgoVersion, FormatV3},
-    disk::DiskStore,
+    disk::{
+        format::{DistributionAlgoVersion, FormatV3},
+        DiskStore,
+    },
     endpoints::PoolEndpoints,
     error::Result,
     set_disk::SetDisks,
     store_api::{
-        BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, MakeBucketOptions, MultipartUploadResult,
-        ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
+        BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsInfo, ListObjectsV2Info,
+        MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
     },
     utils::hash,
 };
@@ -133,6 +135,19 @@ impl StorageAPI for Sets {
         unimplemented!()
     }
 
+    async fn list_objects_v2(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        continuation_token: &str,
+        delimiter: &str,
+        max_keys: i32,
+        fetch_owner: bool,
+        start_after: &str,
+    ) -> Result<ListObjectsV2Info> {
+        unimplemented!()
+    }
+
     async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
         self.get_disks_by_key(object).get_object_info(bucket, object, opts).await
     }

ecstore/src/store.rs

@@ -1,20 +1,22 @@
 use crate::{
     bucket_meta::BucketMetadata,
-    disk::{error::DiskError, DiskOption, DiskStore, RUSTFS_META_BUCKET},
+    disk::{error::DiskError, DiskOption, DiskStore, MetaCacheEntry, WalkDirOptions, RUSTFS_META_BUCKET},
     disks_layout::DisksLayout,
     endpoints::EndpointServerPools,
     error::{Error, Result},
     peer::{PeerS3Client, S3PeerSys},
     sets::Sets,
     store_api::{
-        BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, MakeBucketOptions, MultipartUploadResult,
-        ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
+        BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsInfo, ListObjectsV2Info,
+        MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
     },
     store_init, utils,
 };
+use futures::future::join_all;
 use http::HeaderMap;
 use s3s::{dto::StreamingBlob, Body};
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
+use tokio::sync::mpsc;
 use uuid::Uuid;
 
 #[derive(Debug)]
@@ -105,6 +107,82 @@ impl ECStore {
     fn single_pool(&self) -> bool {
         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 listMerged(&self, opts: &ListPathOptions, results: mpsc::Sender<MetaCacheEntry>) -> 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 futures = Vec::new();
+
+        for sets in self.pools.iter() {
+            for set in sets.disk_set.iter() {
+                for disk in set.disks.iter() {
+                    if disk.is_none() {
+                        continue;
+                    }
+
+                    let disk = disk.as_ref().unwrap();
+                    let opts = opts.clone();
+                    let wr = wr.clone();
+                    futures.push(disk.walk_dir(opts, wr));
+                    // tokio::spawn(async move { disk.walk_dir(opts, wr).await });
+                }
+            }
+        }
+
+        let results = join_all(futures).await;
+
+        let mut errs = Vec::new();
+
+        for res in results {
+            match res {
+                Ok(_) => errs.push(None),
+                Err(e) => errs.push(Some(e)),
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[derive(Debug, Default)]
+struct ListPathOptions {
+    pub id: String,
+
+    // Bucket of the listing.
+    pub bucket: String,
+
+    // Directory inside the bucket.
+    // When unset listPath will set this based on Prefix
+    pub base_dir: String,
+
+    // Scan/return only content with prefix.
+    pub prefix: String,
+
+    // FilterPrefix will return only results with this prefix when scanning.
+    // Should never contain a slash.
+    // Prefix should still be set.
+    pub filter_prefix: String,
+
+    // Marker to resume listing.
+    // The response will be the first entry >= this object name.
+    pub marker: String,
+
+    // Limit the number of results.
+    pub limit: i32,
 }
 
 #[async_trait::async_trait]
@@ -152,6 +230,27 @@ impl StorageAPI for ECStore {
 
         Ok(info)
     }
+
+    async fn list_objects_v2(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        continuation_token: &str,
+        delimiter: &str,
+        max_keys: i32,
+        fetch_owner: bool,
+        start_after: &str,
+    ) -> Result<ListObjectsV2Info> {
+        let opts = ListPathOptions {
+            bucket: bucket.to_string(),
+            limit: max_keys,
+            ..Default::default()
+        };
+
+        self.list_path(&opts).await?;
+
+        Ok(ListObjectsV2Info::default())
+    }
     async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
         let object = utils::path::encode_dir_object(object);
 

ecstore/src/store_api.rs

@@ -424,12 +424,67 @@ pub struct ObjectInfo {
     pub is_latest: bool,
 }
 
+#[derive(Debug)]
+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
+    // if the number of objects exceeds the limit allowed or specified
+    // by max keys.
+    pub is_truncated: bool,
+
+    // When response is truncated (the IsTruncated element value in the response
+    // is true), you can use the key name in this field as marker in the subsequent
+    // request to get next set of objects.
+    pub next_marker: String,
+
+    // List of objects info for this request.
+    pub objects: Vec<ObjectInfo>,
+
+    // List of prefixes for this request.
+    pub prefixes: Vec<String>,
+}
+
+#[derive(Debug, Default)]
+pub struct ListObjectsV2Info {
+    // Indicates whether the returned list objects response is truncated. A
+    // value of true indicates that the list was truncated. The list can be truncated
+    // if the number of objects exceeds the limit allowed or specified
+    // by max keys.
+    pub is_truncated: bool,
+
+    // When response is truncated (the IsTruncated element value in the response
+    // is true), you can use the key name in this field as marker in the subsequent
+    // request to get next set of objects.
+    //
+    // NOTE: This element is returned only if you have delimiter request parameter
+    // specified.
+    pub continuation_token: String,
+    pub next_continuation_token: String,
+
+    // List of objects info for this request.
+    pub objects: Vec<ObjectInfo>,
+
+    // List of prefixes for this request.
+    pub prefixes: Vec<String>,
+}
+
 #[async_trait::async_trait]
 pub trait StorageAPI {
     async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>;
     async fn delete_bucket(&self, bucket: &str) -> Result<()>;
     async fn list_bucket(&self, opts: &BucketOptions) -> Result<Vec<BucketInfo>>;
     async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo>;
+
+    async fn list_objects_v2(
+        &self,
+        bucket: &str,
+        prefix: &str,
+        continuation_token: &str,
+        delimiter: &str,
+        max_keys: i32,
+        fetch_owner: bool,
+        start_after: &str,
+    ) -> Result<ListObjectsV2Info>;
     async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo>;
     async fn get_object_reader(
         &self,

rustfs/src/storage/ecfs.rs

@@ -247,7 +247,30 @@ impl S3 for FS {
 
     #[tracing::instrument(level = "debug", skip(self, req))]
     async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
-        let _input = req.input;
+        let ListObjectsV2Input {
+            bucket,
+            continuation_token,
+            delimiter,
+            fetch_owner,
+            max_keys,
+            prefix,
+            start_after,
+            ..
+        } = req.input;
+
+        let _object_infos = try_!(
+            self.store
+                .list_objects_v2(
+                    &bucket,
+                    &prefix.unwrap_or_default(),
+                    &continuation_token.unwrap_or_default(),
+                    &delimiter.unwrap_or_default(),
+                    max_keys.unwrap_or_default(),
+                    fetch_owner.unwrap_or_default(),
+                    &start_after.unwrap_or_default()
+                )
+                .await
+        );
 
         let output = ListObjectsV2Output { ..Default::default() };
         Ok(S3Response::new(output))