From 73ee65e3a7342c3e3c2140024434fe1c26819993 Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 17 Jul 2024 14:15:22 +0800 Subject: [PATCH] plugin --- ecstore/src/disk.rs | 29 +++++++++--- ecstore/src/disk_api.rs | 6 ++- ecstore/src/sets.rs | 7 ++- ecstore/src/store.rs | 9 ++++ ecstore/src/store_api.rs | 3 +- ecstore/src/writer.rs | 90 ++++++++++++++++++++++++++++++++++++++ rustfs/src/storage/ecfs.rs | 25 ++++++++++- scripts/run.sh | 2 +- 8 files changed, 160 insertions(+), 11 deletions(-) create mode 100644 ecstore/src/writer.rs diff --git a/ecstore/src/disk.rs b/ecstore/src/disk.rs index e83fe70dd..4feeb57ca 100644 --- a/ecstore/src/disk.rs +++ b/ecstore/src/disk.rs @@ -15,7 +15,9 @@ use tracing::{debug, warn}; use uuid::Uuid; use crate::{ - disk_api::{DeleteOptions, DiskAPI, DiskError, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, VolumeInfo}, + disk_api::{ + DeleteOptions, DiskAPI, DiskError, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, VolumeInfo, + }, endpoint::{Endpoint, Endpoints}, file_meta::FileMeta, format::FormatV3, @@ -197,6 +199,8 @@ impl LocalDisk { } pub async fn delete_file(&self, base_path: &PathBuf, delete_path: &PathBuf, recursive: bool, _immediate: bool) -> Result<()> { + debug!("delete_file {:?}", &delete_path); + if is_root_path(base_path) || is_root_path(delete_path) { return Ok(()); } @@ -403,9 +407,9 @@ impl DiskAPI for LocalDisk { } async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { + let src_volume_path = self.get_bucket_path(&src_volume)?; if !skip_access_checks(&src_volume) { - let vol_path = self.get_bucket_path(&src_volume)?; - check_volume_exists(&vol_path).await?; + check_volume_exists(&src_volume_path).await?; } if !skip_access_checks(&dst_volume) { let vol_path = self.get_bucket_path(&dst_volume)?; @@ -440,6 +444,12 @@ impl DiskAPI for LocalDisk { break; } + + if let Some(dir_path) = srcp.parent() { + self.delete_file(&src_volume_path, &PathBuf::from(dir_path), false, false) + .await?; + } + Ok(()) } @@ -492,7 +502,14 @@ impl DiskAPI for LocalDisk { // Ok(()) } - async fn rename_data(&self, src_volume: &str, src_path: &str, fi: &FileInfo, dst_volume: &str, dst_path: &str) -> Result<()> { + async fn rename_data( + &self, + src_volume: &str, + src_path: &str, + fi: &FileInfo, + dst_volume: &str, + dst_path: &str, + ) -> Result { let src_volume_path = self.get_bucket_path(&src_volume)?; if !skip_access_checks(&src_volume) { check_volume_exists(&src_volume_path).await?; @@ -565,7 +582,9 @@ impl DiskAPI for LocalDisk { .await?; } - Ok(()) + Ok(RenameDataResp { + old_data_dir: fi.data_dir.to_string(), + }) } async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { diff --git a/ecstore/src/disk_api.rs b/ecstore/src/disk_api.rs index 7ea8ba344..0f438f3a5 100644 --- a/ecstore/src/disk_api.rs +++ b/ecstore/src/disk_api.rs @@ -26,7 +26,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { file_info: &FileInfo, dst_volume: &str, dst_path: &str, - ) -> Result<()>; + ) -> Result; async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; async fn make_volume(&self, volume: &str) -> Result<()>; @@ -45,6 +45,10 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; } +pub struct RenameDataResp { + pub old_data_dir: String, +} + #[derive(Debug, Clone, Default)] pub struct DeleteOptions { pub recursive: bool, diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index f718de5f8..bdf3d1f82 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -7,8 +7,8 @@ use crate::{ format::{DistributionAlgoVersion, FormatV3}, set_disk::SetDisks, store_api::{ - BucketInfo, BucketOptions, CompletePart, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, PartInfo, - PutObjReader, StorageAPI, + BucketInfo, BucketOptions, CompletePart, FileInfo, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, + PartInfo, PutObjReader, StorageAPI, }, utils::hash, }; @@ -129,6 +129,9 @@ impl StorageAPI for Sets { 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 + } async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: &ObjectOptions) -> Result<()> { self.get_disks_by_key(object).put_object(bucket, object, data, opts).await } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 96061bd6f..9148f9556 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -151,6 +151,15 @@ impl StorageAPI for ECStore { Ok(info) } + async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { + let object = utils::path::encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].get_object_info(bucket, object.as_str(), opts).await; + } + + unimplemented!() + } async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: &ObjectOptions) -> Result<()> { // checkPutObjectArgs diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 59eb2d050..477c9e0fc 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -282,6 +282,7 @@ impl From for CompletePart { } } +#[derive(Debug)] pub struct ObjectInfo { pub bucket: String, pub name: String, @@ -300,7 +301,7 @@ pub struct ObjectInfo { pub trait StorageAPI { async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()>; async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result; - + async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result; async fn put_object(&self, bucket: &str, object: &str, data: PutObjReader, opts: &ObjectOptions) -> Result<()>; async fn put_object_part( &self, diff --git a/ecstore/src/writer.rs b/ecstore/src/writer.rs new file mode 100644 index 000000000..48257dc05 --- /dev/null +++ b/ecstore/src/writer.rs @@ -0,0 +1,90 @@ +use std::{io, task::Poll}; + +use futures::{ready, Future}; +use tokio::io::{AsyncWrite, BufWriter}; +use tracing::debug; +use uuid::Uuid; + +use crate::disk::DiskStore; + +pub struct AppendWriter<'a> { + disk: DiskStore, + volume: &'a str, + path: &'a str, +} + +impl<'a> AppendWriter<'a> { + pub fn new(disk: DiskStore, volume: &'a str, path: &'a str) -> Self { + debug!("AppendWriter new {}: {}/{}", disk.id(), volume, path); + Self { disk, volume, path } + } + + async fn async_write(&self, buf: &[u8]) -> Result<(), std::io::Error> { + debug!("async_write {}: {}: {}", self.disk.id(), &self.path, buf.len()); + + // self.disk + // .append_file(&self.volume, &self.path, buf) + // .await + // .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?; + Ok(()) + } +} + +impl<'a> AsyncWrite for AppendWriter<'a> { + fn poll_write( + self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + let mut fut = Box::pin(self.async_write(buf)); + debug!("AsyncWrite poll_write {}, buf:{}", self.disk.id(), buf.len()); + + // while let Poll::Ready(e) = fut.as_mut().poll(cx) { + // let a = match e { + // Ok(_) => { + // debug!("Ready ok {}", self.disk.id()); + // Poll::Ready(Ok(buf.len())) + // } + // Err(e) => { + // debug!("Ready err {}", self.disk.id()); + // Poll::Ready(Err(e)) + // } + // }; + + // return a; + // } + + // Poll::Pending + + match fut.as_mut().poll(cx) { + Poll::Pending => { + debug!("Pending {}", self.disk.id()); + Poll::Pending + } + Poll::Ready(e) => match e { + Ok(_) => { + debug!("Ready ok {}", self.disk.id()); + Poll::Ready(Ok(buf.len())) + } + Err(e) => { + debug!("Ready err {}", self.disk.id()); + Poll::Ready(Err(e)) + } + }, + } + } + + fn poll_flush( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + Poll::Ready(Ok(())) + } + + fn poll_shutdown( + self: std::pin::Pin<&mut Self>, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + Poll::Ready(Ok(())) + } +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 364e9ae5f..ce59e1157 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -18,6 +18,8 @@ use s3s::{S3Request, S3Response}; use anyhow::Result; use ecstore::store::ECStore; +use tracing::debug; +use tracing::info; macro_rules! try_ { ($result:expr) => { @@ -95,6 +97,7 @@ impl S3 for FS { #[tracing::instrument] async fn get_bucket_location(&self, req: S3Request) -> S3Result> { + // mc get 1 let input = req.input; if let Err(e) = self.store.get_bucket_info(&input.bucket, &BucketOptions {}).await { @@ -146,7 +149,27 @@ impl S3 for FS { #[tracing::instrument] async fn head_object(&self, req: S3Request) -> S3Result> { - let _input = req.input; + // mc get 2 + let HeadObjectInput { + bucket, + checksum_mode, + expected_bucket_owner, + if_match, + if_modified_since, + if_none_match, + if_unmodified_since, + key, + part_number, + range, + request_payer, + sse_customer_algorithm, + sse_customer_key, + sse_customer_key_md5, + version_id, + } = &req.input; + + let info = try_!(self.store.get_object_info(bucket, key, &ObjectOptions::default()).await); + debug!("info {:?}", info); let output = HeadObjectOutput { ..Default::default() }; Ok(S3Response::new(output)) diff --git a/scripts/run.sh b/scripts/run.sh index d0829fe6a..8e9ee7a67 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -9,7 +9,7 @@ if [ -n "$1" ]; then fi if [ -z "$RUST_LOG" ]; then - export RUST_LOG="s3s-rustfs=debug,ecstore=debug,s3s=debug" + export RUST_LOG="rustfs=debug,ecstore=debug,s3s=debug" fi cargo run \