From 25e98cd1e26433bcf99ab8088430bb68f15bcda8 Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 19 Aug 2024 18:02:22 +0800 Subject: [PATCH] init listobject --- TODO.md | 6 +-- ecstore/src/disk/local.rs | 56 +++++++++++++++++-- ecstore/src/disk/mod.rs | 51 ++++++++++++++++-- ecstore/src/sets.rs | 23 ++++++-- ecstore/src/store.rs | 107 +++++++++++++++++++++++++++++++++++-- ecstore/src/store_api.rs | 55 +++++++++++++++++++ rustfs/src/storage/ecfs.rs | 25 ++++++++- 7 files changed, 302 insertions(+), 21 deletions(-) diff --git a/TODO.md b/TODO.md index c9d2bc34f..344f23eb3 100644 --- a/TODO.md +++ b/TODO.md @@ -2,12 +2,12 @@ ## 基础存储 -- [ ] 优化xlmeta, 自定义msg数据结构 +- [x] 优化xlmeta, 自定义msg数据结构 - [ ] 小文件存储到metafile, inlinedata -- [ ] 上传同名文件时，删除旧版本文件 +- [x] 上传同名文件时，删除旧版本文件 - [ ] EC可用读写数量判断 Read/WriteQuorum - [ ] 错误类型判断,程序中判断错误类型，如何统一错误 -- [ ] 优化并发执行 +- [ ] 优化并发执行，边读边取，可中断 - [ ] 抽象出metafile存储 - [ ] 代码优化 diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 7617e7282..a392efe8f 100644 --- a/ecstore/src/disk/local.rs +++ b/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> { + async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { 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> { - unimplemented!() + async fn walk_dir(&self, opts: WalkDirOptions, wr: Arc) -> 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))] diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index eaec0739a..7489cdfe7 100644 --- a/ecstore/src/disk/mod.rs +++ b/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; async fn read_file(&self, volume: &str, path: &str) -> Result; // 读目录下的所有文件、目录 - async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: usize) -> Result>; - // 读目录下的所有xl.meta - async fn walk_dir(&self) -> Result>; + 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 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>; } +#[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, + + // cached contains the metadata if decoded. + cached: Option, + + // 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, diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 7f606a612..25e2735d6 100644 --- a/ecstore/src/sets.rs +++ b/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 { + unimplemented!() + } + async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { self.get_disks_by_key(object).get_object_info(bucket, object, opts).await } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 2707b0fe3..5f12b15e5 100644 --- a/ecstore/src/store.rs +++ b/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) -> 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 { + 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 { let object = utils::path::encode_dir_object(object); diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 4c3a48f11..1b68aae73 100644 --- a/ecstore/src/store_api.rs +++ b/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, + + // List of prefixes for this request. + pub prefixes: Vec, +} + +#[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, + + // List of prefixes for this request. + pub prefixes: Vec, +} + #[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>; async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result; + + 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; async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result; async fn get_object_reader( &self, diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index f6a02de33..c75bd1904 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/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) -> S3Result> { - 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))