gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 08:18:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

merge global_disks

Browse Source
This commit is contained in:
weisd
2024-09-10 17:59:23 +08:00
parent 21c414a5e9 fc533c7b2b
commit 8d1447e08c
15 changed files with 917 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+3 -2
View File
@@ -1563,6 +1563,7 @@ dependencies = [
"tracing-error",
"tracing-subscriber",
"transform-stream",
"uuid",
]
[[package]]
@@ -2242,9 +2243,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
[[package]]
name = "uuid"
version = "1.9.1"
version = "1.10.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5de17fd2f7da591098415cff336e12965a28061ddace43b59cb3c430179c9439"
checksum = "81dfa00651efa65069b0b6b651f4aaa31ba9e3c3ce0137aaad053604ee7e0314"
dependencies = [
"getrandom",
"rand",
Cargo.toml
+1 -1
View File
@@ -4,7 +4,7 @@ members = ["rustfs", "ecstore", "e2e_test", "common/protos"]
[workspace.package]
edition = "2021"
license = "MIT OR Apache-2.0"
license = "Apache-2.0"
repository = "https://github.com/rustfs/rustfs"
rust-version = "1.75"
version = "0.0.1"
TODO.md
+16 -5
View File
@@ -29,16 +29,27 @@
- [x] 提交完成 CompleteMultipartUpload
- [x] 取消上传 AbortMultipartUpload
- [x] 下载 GetObject
- [ ] 删除 DeleteObjects
- [x] 删除 DeleteObjects
- [ ] 版本控制
- [ ] 对象锁
- [ ] 复制 CopyObject
- [ ] 详情 HeadObject
- [ ] 对象预先签名（get、put、head、post）
## 扩展功能
- [ ] 版本控制
- [ ] 对象锁
- [ ] 修复
- [ ] 用户管理
- [ ] Policy管理
- [ ] AK/SK分配管理
- [ ] data scanner统计和对象修复
- [ ] 桶配额
- [ ] 桶只读
- [ ] 桶复制
- [ ] 桶事件通知
- [ ] 桶公开、桶私有
- [ ] 对象生命周期管理
- [ ] prometheus对接
- [ ] 日志收集和日志外发
- [ ] 对象压缩
- [ ] STS
- [ ] 分层（阿里云、腾讯云、S3远程对接）
ecstore/src/disk/local.rs
+101 -31
View File
@@ -1,7 +1,7 @@
use super::{endpoint::Endpoint, error::DiskError, format::FormatV3};
use super::{
DeleteOptions, DiskAPI, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions,
RenameDataResp, VolumeInfo, WalkDirOptions,
DeleteOptions, DiskAPI, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp,
ReadOptions, RenameDataResp, VolumeInfo, WalkDirOptions,
};
use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
use crate::{
@@ -138,8 +138,34 @@ impl LocalDisk {
Ok(())
}
pub async fn move_to_trash(&self, delete_path: &PathBuf, _recursive: bool, _immediate_purge: bool) -> Result<()> {
let trash_path = self.get_object_path(super::RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
// TODO: 清空回收站
if let Err(err) = fs::rename(&delete_path, &trash_path).await {
match err.kind() {
ErrorKind::NotFound => (),
_ => {
warn!("delete_file rename {:?} err {:?}", &delete_path, &err);
return Err(Error::from(err));
}
}
}
// FIXME: 先清空回收站吧，有时间再添加判断逻辑
let _ = fs::remove_dir_all(&trash_path).await;
// TODO: immediate
Ok(())
}
// #[tracing::instrument(skip(self))]
pub async fn delete_file(&self, base_path: &PathBuf, delete_path: &PathBuf, recursive: bool, _immediate: bool) -> Result<()> {
pub async fn delete_file(
&self,
base_path: &PathBuf,
delete_path: &PathBuf,
recursive: bool,
immediate_purge: bool,
) -> Result<()> {
debug!("delete_file {:?}\n base_path:{:?}", &delete_path, &base_path);
if is_root_path(base_path) || is_root_path(delete_path) {
@@ -153,29 +179,7 @@ impl LocalDisk {
}
if recursive {
let trash_path = self.get_object_path(super::RUSTFS_META_TMP_DELETED_BUCKET, Uuid::new_v4().to_string().as_str())?;
if let Some(dir_path) = trash_path.parent() {
fs::create_dir_all(dir_path).await?;
}
debug!("delete_file ranme to trash {:?} to {:?}", &delete_path, &trash_path);
// TODO: 清空回收站
if let Err(err) = fs::rename(&delete_path, &trash_path).await {
match err.kind() {
ErrorKind::NotFound => (),
_ => {
warn!("delete_file rename {:?} err {:?}", &delete_path, &err);
return Err(Error::from(err));
}
}
}
// FIXME: 先清空回收站吧，有时间再添加判断逻辑
let _ = fs::remove_dir_all(&trash_path).await;
// TODO: immediate
self.move_to_trash(delete_path, recursive, immediate_purge).await?;
} else {
if delete_path.is_dir() {
if let Err(err) = fs::remove_dir(&delete_path).await {
@@ -253,6 +257,52 @@ impl LocalDisk {
Ok((data, modtime))
}
async fn delete_versions_internal(&self, volume: &str, path: &str, fis: &Vec<FileInfo>) -> Result<()> {
let volume_dir = self.get_bucket_path(volume)?;
let xlpath = self.get_object_path(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str())?;
let (data, _) = match self.read_all_data(volume, volume_dir.as_path(), &xlpath).await {
Ok(res) => res,
Err(_err) => {
// TODO: check if not found return err
(Vec::new(), OffsetDateTime::UNIX_EPOCH)
}
};
if data.is_empty() {
return Err(Error::new(DiskError::FileNotFound));
}
let mut fm = FileMeta::default();
fm.unmarshal_msg(&data)?;
for fi in fis {
let data_dir = fm.delete_version(fi)?;
if data_dir.is_some() {
let dir_path = self.get_object_path(volume, format!("{}/{}", path, data_dir.unwrap().to_string()).as_str())?;
self.move_to_trash(&dir_path, true, false).await?;
}
}
// 没有版本了，删除xl.meta
if fm.versions.is_empty() {
self.delete_file(&volume_dir, &xlpath, true, false).await?;
return Ok(());
}
// 更新xl.meta
let buf = fm.marshal_msg()?;
self.write_all(volume, format!("{}/{}", path, super::STORAGE_FORMAT_FILE).as_str(), buf)
.await?;
Ok(())
}
}
fn is_root_path(path: impl AsRef<Path>) -> bool {
@@ -606,7 +656,7 @@ impl DiskAPI for LocalDisk {
let (src_data_path, dst_data_path) = {
let mut data_dir = String::new();
if !fi.is_remote() {
data_dir = utils::path::retain_slash(fi.data_dir.to_string().as_str());
data_dir = utils::path::retain_slash(fi.data_dir.unwrap_or(Uuid::nil()).to_string().as_str());
}
if !data_dir.is_empty() {
@@ -645,10 +695,9 @@ impl DiskAPI for LocalDisk {
let old_data_dir = meta
.find_version(fi.version_id)
.map(|(_, version)| {
version.get_data_dir().filter(|data_dir| {
warn!("get data dir {}", &data_dir);
meta.shard_data_dir_count(&fi.version_id, data_dir) == 0
})
version
.get_data_dir()
.filter(|data_dir| meta.shard_data_dir_count(&fi.version_id, &Some(data_dir.clone())) == 0)
})
.unwrap_or_default();
@@ -819,6 +868,27 @@ impl DiskAPI for LocalDisk {
Ok(RawFileInfo { buf })
}
async fn delete_versions(
&self,
volume: &str,
versions: Vec<FileInfoVersions>,
_opts: DeleteOptions,
) -> Result<Vec<Option<Error>>> {
let mut errs = Vec::with_capacity(versions.len());
for _ in 0..versions.len() {
errs.push(None);
}
for (i, ver) in versions.iter().enumerate() {
if let Err(e) = self.delete_versions_internal(volume, ver.name.as_str(), &ver.versions).await {
errs[i] = Some(e);
} else {
errs[i] = None;
}
}
Ok(errs)
}
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
let mut results = Vec::new();
let mut found = 0;
ecstore/src/disk/mod.rs
+23 -1
View File
@@ -14,7 +14,7 @@ const STORAGE_FORMAT_FILE: &str = "xl.meta";
use crate::{
erasure::{ReadAt, Write},
error::Result,
error::{Error, Result},
file_meta::FileMeta,
store_api::{FileInfo, RawFileInfo},
};
@@ -85,9 +85,31 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
opts: &ReadOptions,
) -> Result<FileInfo>;
async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result<RawFileInfo>;
async fn delete_versions(
&self,
volume: &str,
versions: Vec<FileInfoVersions>,
opts: DeleteOptions,
) -> Result<Vec<Option<Error>>>;
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
}
#[derive(Debug, Default, Clone)]
pub struct FileInfoVersions {
// Name of the volume.
pub volume: String,
// Name of the file.
pub name: String,
// Represents the latest mod time of the
// latest version.
pub latest_mod_time: Option<OffsetDateTime>,
pub versions: Vec<FileInfo>,
pub free_versions: Vec<FileInfo>,
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct WalkDirOptions {
// Bucket to scanner
ecstore/src/disk/remote.rs
+21 -19
View File
@@ -20,13 +20,14 @@ use tracing::info;
use uuid::Uuid;
use crate::{
error::Result,
error::{Error, Result},
store_api::{FileInfo, RawFileInfo},
};
use super::{
endpoint::Endpoint, DeleteOptions, DiskAPI, DiskOption, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq,
ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, VolumeInfo, WalkDirOptions,
endpoint::Endpoint, DeleteOptions, DiskAPI, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry,
ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, VolumeInfo,
WalkDirOptions,
};
#[derive(Debug)]
@@ -40,36 +41,28 @@ impl RemoteDisk {
pub async fn new(ep: &Endpoint, _opt: &DiskOption) -> Result<Self> {
let root = fs::canonicalize(ep.url.path()).await?;
Ok(Self {
Ok(Self {
channel: RwLock::new(None),
url: ep.url.clone(),
root
root,
})
}
fn get_client(&self) -> NodeServiceClient<Timeout<Channel>> {
let channel = {
let read_lock = self.channel.read().unwrap();
if let Some(ref channel) = *read_lock {
channel.clone()
} else {
let addr = format!(
"{}://{}:{}",
self.url.scheme(),
self.url.host_str().unwrap(),
self.url.port().unwrap()
);
let addr = format!("{}://{}:{}", self.url.scheme(), self.url.host_str().unwrap(), self.url.port().unwrap());
info!("disk url: {:?}", addr);
let connector = tonic_Endpoint::from_shared(addr).unwrap();
let new_channel = tokio::runtime::Runtime::new()
.unwrap()
.block_on(connector.connect())
.unwrap();
let new_channel = tokio::runtime::Runtime::new().unwrap().block_on(connector.connect()).unwrap();
*self.channel.write().unwrap() = Some(new_channel.clone());
new_channel
}
};
@@ -347,6 +340,15 @@ impl DiskAPI for RemoteDisk {
Ok(raw_file_info)
}
async fn delete_versions(
&self,
_volume: &str,
_versions: Vec<FileInfoVersions>,
_opts: DeleteOptions,
) -> Result<Vec<Option<Error>>> {
unimplemented!()
}
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>> {
let read_multiple_req = serde_json::to_string(&req)?;
let mut client = self.get_client();
ecstore/src/endpoints.rs
+5
View File
@@ -408,6 +408,11 @@ impl AsMut<Vec<PoolEndpoints>> for EndpointServerPools {
}
impl EndpointServerPools {
pub fn from_volumes(server_addr: &str, endpoints: Vec<String>) -> Result<(EndpointServerPools, SetupType)> {
let layouts = DisksLayout::try_from(endpoints.as_slice())?;
Self::create_server_endpoints(server_addr, &layouts)
}
/// validates and creates new endpoints from input args, supports
/// both ellipses and without ellipses transparently.
pub fn create_server_endpoints(server_addr: &str, disks_layout: &DisksLayout) -> Result<(EndpointServerPools, SetupType)> {
ecstore/src/lib.rs
+2 -2
View File
@@ -1,8 +1,8 @@
mod bucket_meta;
mod chunk_stream;
pub mod disk;
mod disks_layout;
mod endpoints;
pub mod disks_layout;
pub mod endpoints;
pub mod erasure;
pub mod error;
mod file_meta;
ecstore/src/sets.rs
+98 -3
View File
@@ -1,3 +1,6 @@
use std::collections::HashMap;
use futures::future::join_all;
use http::HeaderMap;
use uuid::Uuid;
@@ -7,11 +10,11 @@ use crate::{
DiskStore,
},
endpoints::PoolEndpoints,
error::Result,
error::{Error, Result},
set_disk::SetDisks,
store_api::{
BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsV2Info, MakeBucketOptions,
MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
BucketInfo, BucketOptions, CompletePart, DeletedObject, GetObjectReader, HTTPRangeSpec, ListObjectsV2Info,
MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo, PutObjReader, StorageAPI,
},
utils::hash,
};
@@ -110,6 +113,22 @@ impl Sets {
// ) -> Vec<Option<Error>> {
// unimplemented!()
// }
async fn delete_prefix(&self, bucket: &str, object: &str) -> Result<()> {
let mut futures = Vec::new();
let opt = ObjectOptions {
delete_prefix: true,
..Default::default()
};
for set in self.disk_set.iter() {
futures.push(set.delete_object(bucket, object, opt.clone()));
}
let _results = join_all(futures).await;
Ok(())
}
}
// #[derive(Debug)]
@@ -122,6 +141,12 @@ impl Sets {
// pub default_parity_count: usize,
// }
struct DelObj {
// set_idx: usize,
orig_idx: usize,
obj: ObjectToDelete,
}
#[async_trait::async_trait]
impl StorageAPI for Sets {
async fn list_bucket(&self, _opts: &BucketOptions) -> Result<Vec<BucketInfo>> {
@@ -134,7 +159,77 @@ impl StorageAPI for Sets {
async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
unimplemented!()
}
async fn delete_objects(
&self,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> Result<(Vec<DeletedObject>, Vec<Option<Error>>)> {
// 默认返回值
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..objects.len() {
del_errs.push(None)
}
let mut set_obj_map = HashMap::new();
// hash key
let mut i = 0;
for obj in objects.iter() {
let idx = self.get_hashed_set_index(obj.object_name.as_str());
if !set_obj_map.contains_key(&idx) {
set_obj_map.insert(
idx,
vec![DelObj {
// set_idx: idx,
orig_idx: i,
obj: obj.clone(),
}],
);
} else {
if let Some(val) = set_obj_map.get_mut(&idx) {
val.push(DelObj {
// set_idx: idx,
orig_idx: i,
obj: obj.clone(),
});
}
}
i += 1;
}
// TODO: 并发
for (k, v) in set_obj_map {
let disks = self.get_disks(k);
let objs: Vec<ObjectToDelete> = v.iter().map(|v| v.obj.clone()).collect();
let (dobjects, errs) = disks.delete_objects(bucket, objs, opts.clone()).await?;
let mut i = 0;
for err in errs {
let obj = v.get(i).unwrap();
del_errs[obj.orig_idx] = err;
del_objects[obj.orig_idx] = dobjects.get(i).unwrap().clone();
i += 1;
}
}
Ok((del_objects, del_errs))
}
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
if opts.delete_prefix {
self.delete_prefix(bucket, object).await?;
return Ok(ObjectInfo::default());
}
self.get_disks_by_key(object).delete_object(bucket, object, opts).await
}
async fn list_objects_v2(
&self,
_bucket: &str,
ecstore/src/store.rs
+247 -9
View File
@@ -1,24 +1,45 @@
use crate::{
bucket_meta::BucketMetadata,
disk::{error::DiskError, DeleteOptions, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
disks_layout::DisksLayout,
endpoints::EndpointServerPools,
error::{Error, Result},
peer::S3PeerSys,
sets::Sets,
store_api::{
BucketInfo, BucketOptions, CompletePart, GetObjectReader, HTTPRangeSpec, ListObjectsInfo, ListObjectsV2Info,
MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
BucketInfo, BucketOptions, CompletePart, DeletedObject, GetObjectReader, HTTPRangeSpec, ListObjectsInfo,
ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo,
PutObjReader, StorageAPI,
},
store_init, utils,
};
use futures::future::join_all;
use http::HeaderMap;
use s3s::{dto::StreamingBlob, Body};
use std::collections::{HashMap, HashSet};
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use time::OffsetDateTime;
use tokio::sync::Mutex;
use tracing::{debug, warn};
use uuid::Uuid;
use lazy_static::lazy_static;
lazy_static! {
pub static ref GLOBAL_OBJECT_API: Arc<Mutex<Option<ECStore>>> = Arc::new(Mutex::new(None));
}
pub fn new_object_layer_fn() -> Arc<Mutex<Option<ECStore>>> {
// 这里不需要显式地锁定和解锁，因为 Arc 提供了必要的线程安全性
GLOBAL_OBJECT_API.clone()
}
async fn set_object_layer(o: ECStore) {
let mut global_object_api = GLOBAL_OBJECT_API.lock().await;
*global_object_api = Some(o);
}
#[derive(Debug)]
pub struct ECStore {
pub id: uuid::Uuid,
@@ -30,12 +51,12 @@ pub struct ECStore {
}
impl ECStore {
pub async fn new(address: String, endpoints: Vec<String>) -> Result<Self> {
let layouts = DisksLayout::try_from(endpoints.as_slice())?;
pub async fn new(_address: String, endpoint_pools: EndpointServerPools) -> Result<()> {
// let layouts = DisksLayout::try_from(endpoints.as_slice())?;
let mut deployment_id = None;
let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?;
// let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?;
let mut pools = Vec::with_capacity(endpoint_pools.as_ref().len());
let mut disk_map = HashMap::with_capacity(endpoint_pools.as_ref().len());
@@ -95,13 +116,17 @@ impl ECStore {
let peer_sys = S3PeerSys::new(&endpoint_pools, local_disks.clone());
Ok(ECStore {
let ec = ECStore {
id: deployment_id.unwrap(),
disk_map,
pools,
local_disks,
peer_sys,
})
};
set_object_layer(ec).await;
Ok(())
}
pub fn local_disks(&self) -> Vec<DiskStore> {
@@ -231,6 +256,77 @@ impl ECStore {
Ok(())
}
async fn delete_prefix(&self, _bucket: &str, _object: &str) -> Result<()> {
unimplemented!()
}
async fn get_pool_info_existing_with_opts(
&self,
bucket: &str,
object: &str,
opts: &ObjectOptions,
) -> Result<(PoolObjInfo, Vec<Error>)> {
let mut futures = Vec::new();
for pool in self.pools.iter() {
futures.push(pool.get_object_info(bucket, object, opts));
}
let results = join_all(futures).await;
let mut ress = Vec::new();
let mut i = 0;
// join_all结果跟输入顺序一致
for res in results {
let index = i;
match res {
Ok(r) => {
ress.push(PoolObjInfo {
index,
object_info: r,
err: None,
});
}
Err(e) => {
ress.push(PoolObjInfo {
index,
err: Some(e),
..Default::default()
});
}
}
i += 1;
}
ress.sort_by(|a, b| {
let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
at.cmp(&bt)
});
for res in ress {
// check
if res.err.is_none() {
// TODO: let errs = self.poolsWithObject()
return Ok((res, Vec::new()));
}
}
let ret = PoolObjInfo::default();
Ok((ret, Vec::new()))
}
}
#[derive(Debug, Default)]
pub struct PoolObjInfo {
pub index: usize,
pub object_info: ObjectInfo,
pub err: Option<Error>,
}
#[derive(Debug, Default)]
@@ -305,7 +401,149 @@ impl StorageAPI for ECStore {
Ok(info)
}
async fn delete_objects(
&self,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> Result<(Vec<DeletedObject>, Vec<Option<Error>>)> {
// encode object name
let objects: Vec<ObjectToDelete> = objects
.iter()
.map(|v| {
let mut v = v.clone();
v.object_name = utils::path::encode_dir_object(v.object_name.as_str());
v
})
.collect();
// 默认返回值
let mut del_objects = vec![DeletedObject::default(); objects.len()];
let mut del_errs = Vec::with_capacity(objects.len());
for _ in 0..objects.len() {
del_errs.push(None)
}
// TODO: limte 限制并发数量
let opt = ObjectOptions::default();
// 取所有poolObjInfo
let mut futures = Vec::new();
for obj in objects.iter() {
futures.push(self.get_pool_info_existing_with_opts(bucket, &obj.object_name, &opt));
}
let results = join_all(futures).await;
// 记录pool Index 对应的objects pool_idx -> objects idx
let mut pool_index_objects = HashMap::new();
let mut i = 0;
for res in results {
match res {
Ok((pinfo, _)) => {
if pinfo.object_info.delete_marker && opts.version_id.is_empty() {
del_objects[i] = DeletedObject {
delete_marker: pinfo.object_info.delete_marker,
delete_marker_version_id: pinfo.object_info.version_id.map(|v| v.to_string()),
object_name: utils::path::decode_dir_object(&pinfo.object_info.name),
delete_marker_mtime: pinfo.object_info.mod_time,
..Default::default()
};
}
if !pool_index_objects.contains_key(&pinfo.index) {
pool_index_objects.insert(pinfo.index, vec![i]);
} else {
// let mut vals = pool_index_objects.
if let Some(val) = pool_index_objects.get_mut(&pinfo.index) {
val.push(i);
}
}
}
Err(e) => {
//TODO: check not found
del_errs[i] = Some(e)
}
}
i += 1;
}
if !pool_index_objects.is_empty() {
for sets in self.pools.iter() {
// 取pool idx 对应的 objects index
let vals = pool_index_objects.get(&sets.pool_idx);
if vals.is_none() {
continue;
}
let obj_idxs = vals.unwrap();
// 取对应obj,理论上不会none
let objs: Vec<ObjectToDelete> = obj_idxs
.iter()
.filter_map(|&idx| {
if let Some(obj) = objects.get(idx) {
Some(obj.clone())
} else {
None
}
})
.collect();
if objs.is_empty() {
continue;
}
let (pdel_objs, perrs) = sets.delete_objects(bucket, objs, opts.clone()).await?;
// perrs的顺序理论上跟obj_idxs顺序一致
let mut i = 0;
for err in perrs {
let obj_idx = obj_idxs[i];
if err.is_some() {
del_errs[obj_idx] = err;
}
let mut dobj = pdel_objs.get(i).unwrap().clone();
dobj.object_name = utils::path::decode_dir_object(&dobj.object_name);
del_objects[obj_idx] = dobj;
i += 1;
}
}
}
Ok((del_objects, del_errs))
}
async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
if opts.delete_prefix {
self.delete_prefix(bucket, &object).await?;
return Ok(ObjectInfo::default());
}
let object = utils::path::encode_dir_object(object);
let object = object.as_str();
// 查询在哪个pool
let (mut pinfo, errs) = self.get_pool_info_existing_with_opts(bucket, object, &opts).await?;
if pinfo.object_info.delete_marker && opts.version_id.is_empty() {
pinfo.object_info.name = utils::path::decode_dir_object(object);
return Ok(pinfo.object_info);
}
if !errs.is_empty() {
// TODO: deleteObjectFromAllPools
}
let mut obj = self.pools[pinfo.index].delete_object(bucket, object, opts.clone()).await?;
obj.name = utils::path::decode_dir_object(object);
Ok(obj)
}
async fn list_objects_v2(
&self,
bucket: &str,
ecstore/src/store_api.rs
+95 -72
View File
@@ -10,15 +10,15 @@ pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
pub const BLOCK_SIZE_V2: usize = 1048576; // 1M
// #[derive(Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone)]
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
pub struct FileInfo {
pub name: String,
pub volume: String,
pub version_id: Uuid,
pub version_id: Option<Uuid>,
pub erasure: ErasureInfo,
pub deleted: bool,
// DataDir of the file
pub data_dir: Uuid,
pub data_dir: Option<Uuid>,
pub mod_time: Option<OffsetDateTime>,
pub size: usize,
pub data: Option<Vec<u8>>,
@@ -27,7 +27,66 @@ pub struct FileInfo {
pub is_latest: bool,
}
// impl Default for FileInfo {
// fn default() -> Self {
// Self {
// version_id: Default::default(),
// erasure: Default::default(),
// deleted: Default::default(),
// data_dir: Default::default(),
// mod_time: None,
// size: Default::default(),
// data: Default::default(),
// fresh: Default::default(),
// name: Default::default(),
// volume: Default::default(),
// parts: Default::default(),
// is_latest: Default::default(),
// }
// }
// }
impl FileInfo {
pub fn new(object: &str, data_blocks: usize, parity_blocks: usize) -> Self {
let indexs = {
let cardinality = data_blocks + parity_blocks;
let mut nums = vec![0; cardinality];
let key_crc = crc32fast::hash(object.as_bytes());
let start = key_crc as usize % cardinality;
for i in 1..=cardinality {
nums[i - 1] = 1 + ((start + i) % cardinality);
}
nums
};
Self {
erasure: ErasureInfo {
algorithm: String::from(ERASURE_ALGORITHM),
data_blocks,
parity_blocks,
block_size: BLOCK_SIZE_V2,
distribution: indexs,
..Default::default()
},
..Default::default()
}
}
pub fn is_valid(&self) -> bool {
if self.deleted {
return true;
}
let data_blocks = self.erasure.data_blocks;
let parity_blocks = self.erasure.parity_blocks;
(data_blocks >= parity_blocks)
&& (data_blocks > 0)
&& (self.erasure.index > 0
&& self.erasure.index <= data_blocks + parity_blocks
&& self.erasure.distribution.len() == (data_blocks + parity_blocks))
}
pub fn is_remote(&self) -> bool {
// TODO: when lifecycle
false
@@ -86,7 +145,7 @@ impl FileInfo {
parity_blocks: self.erasure.parity_blocks,
data_blocks: self.erasure.data_blocks,
version_id: self.version_id,
deleted: self.deleted,
delete_marker: self.deleted,
mod_time: self.mod_time,
size: self.size,
parts: self.parts.clone(),
@@ -113,68 +172,6 @@ impl FileInfo {
}
}
impl Default for FileInfo {
fn default() -> Self {
Self {
version_id: Uuid::nil(),
erasure: Default::default(),
deleted: Default::default(),
data_dir: Uuid::nil(),
mod_time: None,
size: Default::default(),
data: Default::default(),
fresh: Default::default(),
name: Default::default(),
volume: Default::default(),
parts: Default::default(),
is_latest: Default::default(),
}
}
}
impl FileInfo {
pub fn new(object: &str, data_blocks: usize, parity_blocks: usize) -> Self {
let indexs = {
let cardinality = data_blocks + parity_blocks;
let mut nums = vec![0; cardinality];
let key_crc = crc32fast::hash(object.as_bytes());
let start = key_crc as usize % cardinality;
for i in 1..=cardinality {
nums[i - 1] = 1 + ((start + i) % cardinality);
}
nums
};
Self {
erasure: ErasureInfo {
algorithm: String::from(ERASURE_ALGORITHM),
data_blocks,
parity_blocks,
block_size: BLOCK_SIZE_V2,
distribution: indexs,
..Default::default()
},
..Default::default()
}
}
pub fn is_valid(&self) -> bool {
if self.deleted {
return true;
}
let data_blocks = self.erasure.data_blocks;
let parity_blocks = self.erasure.parity_blocks;
(data_blocks >= parity_blocks)
&& (data_blocks > 0)
&& (self.erasure.index > 0
&& self.erasure.index <= data_blocks + parity_blocks
&& self.erasure.distribution.len() == (data_blocks + parity_blocks))
}
}
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
pub struct ObjectPartInfo {
// pub etag: Option<String>,
@@ -360,12 +357,15 @@ impl HTTPRangeSpec {
}
}
#[derive(Debug, Default)]
#[derive(Debug, Default, Clone)]
pub struct ObjectOptions {
// Use the maximum parity (N/2), used when saving server configuration files
pub max_parity: bool,
pub mod_time: Option<OffsetDateTime>,
pub part_number: usize,
pub delete_prefix: bool,
pub version_id: String,
}
// impl Default for ObjectOptions {
@@ -416,13 +416,13 @@ impl From<s3s::dto::CompletedPart> for CompletePart {
pub struct ObjectInfo {
pub bucket: String,
pub name: String,
pub mod_time: Option<OffsetDateTime>,
pub size: usize,
pub is_dir: bool,
pub parity_blocks: usize,
pub data_blocks: usize,
pub version_id: Uuid,
pub deleted: bool,
pub mod_time: Option<OffsetDateTime>,
pub size: usize,
pub version_id: Option<Uuid>,
pub delete_marker: bool,
pub parts: Vec<ObjectPartInfo>,
pub is_latest: bool,
}
@@ -471,13 +471,36 @@ pub struct ListObjectsV2Info {
pub prefixes: Vec<String>,
}
#[derive(Debug, Default, Clone)]
pub struct ObjectToDelete {
pub object_name: String,
pub version_id: Option<Uuid>,
}
#[derive(Debug, Default, Clone)]
pub struct DeletedObject {
pub delete_marker: bool,
pub delete_marker_version_id: Option<String>,
pub object_name: String,
pub version_id: Option<String>,
// MTime of DeleteMarker on source that needs to be propagated to replica
pub delete_marker_mtime: Option<OffsetDateTime>,
// to support delete marker replication
// pub replication_state: ReplicationState,
}
#[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 delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo>;
async fn delete_objects(
&self,
bucket: &str,
objects: Vec<ObjectToDelete>,
opts: ObjectOptions,
) -> Result<(Vec<DeletedObject>, Vec<Option<Error>>)>;
async fn list_objects_v2(
&self,
bucket: &str,
rustfs/Cargo.toml
+20 -1
View File
@@ -43,7 +43,26 @@ tower.workspace = true
tracing-error.workspace = true
tracing-subscriber.workspace = true
transform-stream.workspace = true
uuid = "1.10.0"
[build-dependencies]
prost-build.workspace = true
tonic-build.workspace = true
tonic-build.workspace = true
http.workspace = true
bytes.workspace = true
futures.workspace = true
futures-util.workspace = true
# uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] }
ecstore = { path = "../ecstore" }
s3s = "0.10.0"
clap = { version = "4.5.7", features = ["derive"] }
tracing-subscriber = { version = "0.3.18", features = ["env-filter", "time"] }
hyper-util = { version = "0.1.5", features = [
"tokio",
"server-auto",
"server-graceful",
] }
mime = "0.3.17"
transform-stream = "0.3.0"
netif = "0.1.6"
# pin-utils = "0.1.0"
rustfs/src/grpc.rs
+4 -1
View File
@@ -1,5 +1,6 @@
use ecstore::{
disk::{DeleteOptions, DiskStore, ReadMultipleReq, ReadOptions, WalkDirOptions},
endpoints::EndpointServerPools,
erasure::{ReadAt, Write},
peer::{LocalPeerS3Client, PeerS3Client},
store_api::{BucketOptions, FileInfo, MakeBucketOptions},
@@ -28,7 +29,9 @@ struct NodeService {
pub local_peer: LocalPeerS3Client,
}
pub fn make_server(local_disks: Vec<DiskStore>) -> NodeServer<impl Node> {
pub fn make_server(endpoint_pools: EndpointServerPools) -> NodeServer<impl Node> {
// TODO: 参考rustfs创建 https://github.com/rustfs/s3-rustfs/issues/30#issuecomment-2339664516
let local_disks = Vec::new();
let local_peer = LocalPeerS3Client::new(local_disks, None, None);
NodeServer::new(NodeService { local_peer })
}
rustfs/src/main.rs
+58 -40
View File
@@ -4,7 +4,7 @@ mod service;
mod storage;
use clap::Parser;
use ecstore::{error::Result, store::ECStore};
use ecstore::{endpoints::EndpointServerPools, error::Result, store::ECStore};
use grpc::make_server;
use hyper_util::{
rt::{TokioExecutor, TokioIo},
@@ -15,7 +15,7 @@ use s3s::{auth::SimpleAuth, service::S3ServiceBuilder};
use service::hybrid;
use std::{io::IsTerminal, net::SocketAddr, str::FromStr};
use tokio::net::TcpListener;
use tracing::{debug, info};
use tracing::{debug, info, warn};
use tracing_error::ErrorLayer;
use tracing_subscriber::{fmt, layer::SubscriberExt, util::SubscriberInitExt};
@@ -36,10 +36,13 @@ fn setup_tracing() {
}
fn main() -> Result<()> {
//解析获得到的参数
let opt = config::Opt::parse();
//设置trace
setup_tracing();
//运行参数
run(opt)
}
@@ -47,7 +50,9 @@ fn main() -> Result<()> {
async fn run(opt: config::Opt) -> Result<()> {
debug!("opt: {:?}", &opt);
//监听地址,端口从参数中获取
let listener = TcpListener::bind(opt.address.clone()).await?;
//获取监听地址
let local_addr: SocketAddr = listener.local_addr()?;
// let mut domain_name = {
@@ -66,13 +71,15 @@ async fn run(opt: config::Opt) -> Result<()> {
// })
// };
let store: ECStore = ECStore::new(opt.address.clone(), opt.volumes.clone()).await?;
let local_disks = store.local_disks();
// 用于rpc
let (endpoint_pools, _) = EndpointServerPools::from_volumes(opt.address.clone().as_str(), opt.volumes.clone())?;
// Setup S3 service
// 本项目使用s3s库来实现s3服务
let service = {
let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(store).await?);
// let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new(opt.address.clone(), endpoint_pools).await?);
let mut b = S3ServiceBuilder::new(storage::ecfs::FS::new());
//设置AK和SK
//其中部份内容从config配置文件中读取
let mut access_key = String::from_str(config::DEFAULT_ACCESS_KEY).unwrap();
let mut secret_key = String::from_str(config::DEFAULT_SECRET_KEY).unwrap();
@@ -81,7 +88,7 @@ async fn run(opt: config::Opt) -> Result<()> {
access_key = ak;
secret_key = sk;
}
//显示info信息
info!("authentication is enabled {}, {}", &access_key, &secret_key);
b.set_auth(SimpleAuth::from_single(access_key, secret_key));
@@ -101,47 +108,58 @@ async fn run(opt: config::Opt) -> Result<()> {
b.build()
};
let rpc_service = make_server(endpoint_pools.clone());
let hyper_service = service.into_shared();
tokio::spawn(async move {
let hyper_service = service.into_shared();
let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, make_server(local_disks)));
let hybrid_service = TowerToHyperService::new(hybrid(hyper_service, rpc_service));
let http_server = ConnBuilder::new(TokioExecutor::new());
let graceful = hyper_util::server::graceful::GracefulShutdown::new();
let http_server = ConnBuilder::new(TokioExecutor::new());
let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
let graceful = hyper_util::server::graceful::GracefulShutdown::new();
info!("server is running at http://{local_addr}");
let mut ctrl_c = std::pin::pin!(tokio::signal::ctrl_c());
info!("server is running at http://{local_addr}");
loop {
let (socket, _) = tokio::select! {
res = listener.accept() => {
match res {
Ok(conn) => conn,
Err(err) => {
tracing::error!("error accepting connection: {err}");
continue;
loop {
let (socket, _) = tokio::select! {
res = listener.accept() => {
match res {
Ok(conn) => conn,
Err(err) => {
tracing::error!("error accepting connection: {err}");
continue;
}
}
}
}
_ = ctrl_c.as_mut() => {
break;
}
};
_ = ctrl_c.as_mut() => {
break;
}
};
let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
let conn = graceful.watch(conn.into_owned());
tokio::spawn(async move {
let _ = conn.await;
});
}
let conn = http_server.serve_connection(TokioIo::new(socket), hybrid_service.clone());
let conn = graceful.watch(conn.into_owned());
tokio::spawn(async move {
let _ = conn.await;
});
}
tokio::select! {
() = graceful.shutdown() => {
tracing::debug!("Gracefully shutdown!");
},
() = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
tracing::debug!("Waited 10 seconds for graceful shutdown, aborting...");
}
}
});
warn!(" init store");
// init store
ECStore::new(opt.address.clone(), endpoint_pools.clone()).await?;
tokio::select! {
() = graceful.shutdown() => {
tracing::debug!("Gracefully shutdown!");
},
() = tokio::time::sleep(std::time::Duration::from_secs(10)) => {
tracing::debug!("Waited 10 seconds for graceful shutdown, aborting...");
_ = tokio::signal::ctrl_c() => {
}
}
rustfs/src/storage/ecfs.rs
+223 -34
View File
@@ -1,11 +1,13 @@
use bytes::Bytes;
use ecstore::disk::error::DiskError;
use ecstore::store::new_object_layer_fn;
use ecstore::store_api::BucketOptions;
use ecstore::store_api::CompletePart;
use ecstore::store_api::HTTPRangeSpec;
use ecstore::store_api::MakeBucketOptions;
use ecstore::store_api::MultipartUploadResult;
use ecstore::store_api::ObjectOptions;
use ecstore::store_api::ObjectToDelete;
use ecstore::store_api::PutObjReader;
use ecstore::store_api::StorageAPI;
use futures::pin_mut;
@@ -21,9 +23,9 @@ use s3s::{S3Request, S3Response};
use std::fmt::Debug;
use std::str::FromStr;
use transform_stream::AsyncTryStream;
use uuid::Uuid;
use ecstore::error::Result;
use ecstore::store::ECStore;
use tracing::debug;
macro_rules! try_ {
@@ -39,12 +41,13 @@ macro_rules! try_ {
#[derive(Debug)]
pub struct FS {
pub store: ECStore,
// pub store: ECStore,
}
impl FS {
pub async fn new(store: ECStore) -> Result<Self> {
Ok(Self { store })
pub fn new() -> Self {
// let store: ECStore = ECStore::new(address, endpoint_pools).await?;
Self {}
}
}
#[async_trait::async_trait]
@@ -57,8 +60,15 @@ impl S3 for FS {
async fn create_bucket(&self, req: S3Request<CreateBucketInput>) -> S3Result<S3Response<CreateBucketOutput>> {
let input = req.input;
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
try_!(
self.store
store
.make_bucket(&input.bucket, &MakeBucketOptions { force_create: true })
.await
);
@@ -83,24 +93,137 @@ impl S3 for FS {
async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
let input = req.input;
try_!(self.store.delete_bucket(&input.bucket).await);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
try_!(store.delete_bucket(&input.bucket).await);
Ok(S3Response::new(DeleteBucketOutput {}))
}
#[tracing::instrument(level = "debug", skip(self, req))]
async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
let _input = req.input;
let DeleteObjectInput {
bucket, key, version_id, ..
} = req.input;
let output = DeleteObjectOutput::default();
let version_id = version_id
.as_ref()
.map(|v| match Uuid::parse_str(v) {
Ok(id) => Some(id),
Err(_) => None,
})
.unwrap_or_default();
let dobj = ObjectToDelete {
object_name: key,
version_id,
};
let objects: Vec<ObjectToDelete> = vec![dobj];
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let (dobjs, _errs) = try_!(store.delete_objects(&bucket, objects, ObjectOptions::default()).await);
// TODO: let errors;
let (delete_marker, version_id) = {
if let Some((a, b)) = dobjs
.iter()
.map(|v| {
let delete_marker = {
if v.delete_marker {
Some(true)
} else {
None
}
};
let version_id = v.version_id.clone();
(delete_marker, version_id)
})
.next()
{
(a, b)
} else {
(None, None)
}
};
let output = DeleteObjectOutput {
delete_marker,
version_id,
..Default::default()
};
Ok(S3Response::new(output))
}
#[tracing::instrument(level = "debug", skip(self, req))]
async fn delete_objects(&self, req: S3Request<DeleteObjectsInput>) -> S3Result<S3Response<DeleteObjectsOutput>> {
let _input = req.input;
// info!("delete_objects args {:?}", req.input);
let output = DeleteObjectsOutput { ..Default::default() };
let DeleteObjectsInput { bucket, delete, .. } = req.input;
let objects: Vec<ObjectToDelete> = delete
.objects
.iter()
.map(|v| {
let version_id = v
.version_id
.as_ref()
.map(|v| match Uuid::parse_str(v) {
Ok(id) => Some(id),
Err(_) => None,
})
.unwrap_or_default();
ObjectToDelete {
object_name: v.key.clone(),
version_id: version_id,
}
})
.collect();
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let (dobjs, _errs) = try_!(store.delete_objects(&bucket, objects, ObjectOptions::default()).await);
// info!("delete_objects res {:?} {:?}", &dobjs, errs);
let deleted = dobjs
.iter()
.map(|v| DeletedObject {
delete_marker: {
if v.delete_marker {
Some(true)
} else {
None
}
},
delete_marker_version_id: v.delete_marker_version_id.clone(),
key: Some(v.object_name.clone()),
version_id: v.version_id.clone(),
})
.collect();
// TODO: let errors;
let output = DeleteObjectsOutput {
deleted: Some(deleted),
// errors,
..Default::default()
};
Ok(S3Response::new(output))
}
@@ -109,7 +232,14 @@ impl S3 for FS {
// mc get 1
let input = req.input;
if let Err(e) = self.store.get_bucket_info(&input.bucket, &BucketOptions {}).await {
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
if let Err(e) = store.get_bucket_info(&input.bucket, &BucketOptions {}).await {
if DiskError::VolumeNotFound.is(&e) {
return Err(s3_error!(NoSuchBucket));
} else {
@@ -145,11 +275,14 @@ impl S3 for FS {
let h = HeaderMap::new();
let opts = &ObjectOptions::default();
let reader = try_!(
self.store
.get_object_reader(bucket.as_str(), key.as_str(), range, h, opts)
.await
);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let reader = try_!(store.get_object_reader(bucket.as_str(), key.as_str(), range, h, opts).await);
let info = reader.object_info;
@@ -172,7 +305,14 @@ impl S3 for FS {
async fn head_bucket(&self, req: S3Request<HeadBucketInput>) -> S3Result<S3Response<HeadBucketOutput>> {
let input = req.input;
if let Err(e) = self.store.get_bucket_info(&input.bucket, &BucketOptions {}).await {
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
if let Err(e) = store.get_bucket_info(&input.bucket, &BucketOptions {}).await {
if DiskError::VolumeNotFound.is(&e) {
return Err(s3_error!(NoSuchBucket));
} else {
@@ -189,7 +329,14 @@ impl S3 for FS {
// mc get 2
let HeadObjectInput { bucket, key, .. } = req.input;
let info = try_!(self.store.get_object_info(&bucket, &key, &ObjectOptions::default()).await);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let info = try_!(store.get_object_info(&bucket, &key, &ObjectOptions::default()).await);
debug!("info {:?}", info);
let content_type = try_!(ContentType::from_str("application/x-msdownload"));
@@ -209,7 +356,14 @@ impl S3 for FS {
async fn list_buckets(&self, _: S3Request<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
// mc ls
let bucket_infos = try_!(self.store.list_bucket(&BucketOptions {}).await);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let bucket_infos = try_!(store.list_bucket(&BucketOptions {}).await);
let buckets: Vec<Bucket> = bucket_infos
.iter()
@@ -257,8 +411,15 @@ impl S3 for FS {
let prefix = prefix.unwrap_or_default();
let delimiter = delimiter.unwrap_or_default();
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let object_infos = try_!(
self.store
store
.list_objects_v2(
&bucket,
&prefix,
@@ -337,9 +498,16 @@ impl S3 for FS {
let reader = PutObjReader::new(body, content_length as usize);
try_!(self.store.put_object(&bucket, &key, reader, &ObjectOptions::default()).await);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
// self.store.put_object(bucket, object, data, opts);
try_!(store.put_object(&bucket, &key, reader, &ObjectOptions::default()).await);
// store.put_object(bucket, object, data, opts);
let output = PutObjectOutput { ..Default::default() };
Ok(S3Response::new(output))
@@ -358,11 +526,15 @@ impl S3 for FS {
debug!("create_multipart_upload meta {:?}", &metadata);
let MultipartUploadResult { upload_id, .. } = try_!(
self.store
.new_multipart_upload(&bucket, &key, &ObjectOptions::default())
.await
);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let MultipartUploadResult { upload_id, .. } =
try_!(store.new_multipart_upload(&bucket, &key, &ObjectOptions::default()).await);
let output = CreateMultipartUploadOutput {
bucket: Some(bucket),
@@ -397,11 +569,14 @@ impl S3 for FS {
let data = PutObjReader::new(body, content_length as usize);
let opts = ObjectOptions::default();
try_!(
self.store
.put_object_part(&bucket, &key, &upload_id, part_id, data, &opts)
.await
);
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
try_!(store.put_object_part(&bucket, &key, &upload_id, part_id, data, &opts).await);
let output = UploadPartOutput { ..Default::default() };
Ok(S3Response::new(output))
@@ -456,8 +631,15 @@ impl S3 for FS {
uploaded_parts.push(CompletePart::from(part));
}
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
try_!(
self.store
store
.complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, opts)
.await
);
@@ -479,9 +661,16 @@ impl S3 for FS {
bucket, key, upload_id, ..
} = req.input;
let layer = new_object_layer_fn();
let lock = layer.lock().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(S3Error::with_message(S3ErrorCode::InternalError, format!("Not init",))),
};
let opts = &ObjectOptions::default();
try_!(
self.store
store
.abort_multipart_upload(bucket.as_str(), key.as_str(), upload_id.as_str(), opts)
.await
);