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

Merge pull request #184 from rustfs/list-objects

Browse Source 
List objects
This commit is contained in:
weisd
2024-12-28 22:34:15 +08:00
committed by GitHub
parent 25fb79776b b4da1006e9
commit 5e131c9b49
33 changed files with 3680 additions and 989 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -731,6 +731,7 @@ version = "0.0.1"
dependencies = [
"ecstore",
"flatbuffers",
"futures",
"lazy_static",
"lock",
"madmin",
common/protos/src/generated/proto_gen/node_service.rs
+17 -12
View File
@@ -300,17 +300,17 @@ pub struct WalkDirRequest {
/// indicate which one in the disks
#[prost(string, tag = "1")]
pub disk: ::prost::alloc::string::String,
#[prost(string, tag = "2")]
pub walk_dir_options: ::prost::alloc::string::String,
#[prost(bytes = "vec", tag = "2")]
pub walk_dir_options: ::prost::alloc::vec::Vec<u8>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct WalkDirResponse {
#[prost(bool, tag = "1")]
pub success: bool,
#[prost(string, repeated, tag = "2")]
pub meta_cache_entry: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
#[prost(message, optional, tag = "3")]
pub error: ::core::option::Option<Error>,
#[prost(string, tag = "2")]
pub meta_cache_entry: ::prost::alloc::string::String,
#[prost(string, optional, tag = "3")]
pub error_info: ::core::option::Option<::prost::alloc::string::String>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct RenameDataRequest {
@@ -1390,7 +1390,7 @@ pub mod node_service_client {
pub async fn walk_dir(
&mut self,
request: impl tonic::IntoRequest<super::WalkDirRequest>,
) -> std::result::Result<tonic::Response<super::WalkDirResponse>, tonic::Status> {
) -> std::result::Result<tonic::Response<tonic::codec::Streaming<super::WalkDirResponse>>, tonic::Status> {
self.inner
.ready()
.await
@@ -1400,7 +1400,7 @@ pub mod node_service_client {
let mut req = request.into_request();
req.extensions_mut()
.insert(GrpcMethod::new("node_service.NodeService", "WalkDir"));
self.inner.unary(req, path, codec).await
self.inner.server_streaming(req, path, codec).await
}
pub async fn rename_data(
&mut self,
@@ -2361,10 +2361,14 @@ pub mod node_service_server {
&self,
request: tonic::Request<super::ListDirRequest>,
) -> std::result::Result<tonic::Response<super::ListDirResponse>, tonic::Status>;
/// Server streaming response type for the WalkDir method.
type WalkDirStream: tonic::codegen::tokio_stream::Stream<Item = std::result::Result<super::WalkDirResponse, tonic::Status>>
+ std::marker::Send
+ 'static;
async fn walk_dir(
&self,
request: tonic::Request<super::WalkDirRequest>,
) -> std::result::Result<tonic::Response<super::WalkDirResponse>, tonic::Status>;
) -> std::result::Result<tonic::Response<Self::WalkDirStream>, tonic::Status>;
async fn rename_data(
&self,
request: tonic::Request<super::RenameDataRequest>,
@@ -3155,9 +3159,10 @@ pub mod node_service_server {
"/node_service.NodeService/WalkDir" => {
#[allow(non_camel_case_types)]
struct WalkDirSvc<T: NodeService>(pub Arc<T>);
impl<T: NodeService> tonic::server::UnaryService<super::WalkDirRequest> for WalkDirSvc<T> {
impl<T: NodeService> tonic::server::ServerStreamingService<super::WalkDirRequest> for WalkDirSvc<T> {
type Response = super::WalkDirResponse;
type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
type ResponseStream = T::WalkDirStream;
type Future = BoxFuture<tonic::Response<Self::ResponseStream>, tonic::Status>;
fn call(&mut self, request: tonic::Request<super::WalkDirRequest>) -> Self::Future {
let inner = Arc::clone(&self.0);
let fut = async move { <T as NodeService>::walk_dir(&inner, request).await };
@@ -3175,7 +3180,7 @@ pub mod node_service_server {
let mut grpc = tonic::server::Grpc::new(codec)
.apply_compression_config(accept_compression_encodings, send_compression_encodings)
.apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
let res = grpc.unary(method, req).await;
let res = grpc.server_streaming(method, req).await;
Ok(res)
};
Box::pin(fut)
common/protos/src/node.proto
+4 -4
View File
@@ -209,13 +209,13 @@ message ListDirResponse {
message WalkDirRequest {
string disk = 1; // indicate which one in the disks
string walk_dir_options = 2;
bytes walk_dir_options = 2;
}
message WalkDirResponse {
bool success = 1;
repeated string meta_cache_entry = 2;
optional Error error = 3;
string meta_cache_entry = 2;
optional string error_info = 3;
}
message RenameDataRequest {
@@ -762,7 +762,7 @@ service NodeService {
// rpc Append(AppendRequest) returns (AppendResponse) {};
rpc ReadAt(stream ReadAtRequest) returns (stream ReadAtResponse) {};
rpc ListDir(ListDirRequest) returns (ListDirResponse) {};
rpc WalkDir(WalkDirRequest) returns (WalkDirResponse) {};
rpc WalkDir(WalkDirRequest) returns (stream WalkDirResponse) {};
rpc RenameData(RenameDataRequest) returns (RenameDataResponse) {};
rpc MakeVolumes(MakeVolumesRequest) returns (MakeVolumesResponse) {};
rpc MakeVolume(MakeVolumeRequest) returns (MakeVolumeResponse) {};
e2e_test/Cargo.toml
+1
View File
@@ -14,6 +14,7 @@ workspace = true
[dependencies]
ecstore.workspace = true
flatbuffers.workspace = true
futures.workspace = true
lazy_static.workspace = true
lock.workspace = true
protos.workspace = true
e2e_test/src/reliant/node_interact_test.rs
+67 -3
View File
@@ -1,6 +1,9 @@
#![cfg(test)]
use ecstore::disk::VolumeInfo;
use ecstore::disk::{MetaCacheEntry, VolumeInfo, WalkDirOptions};
use ecstore::metacache::writer::{MetacacheReader, MetacacheWriter};
use futures::future::join_all;
use protos::proto_gen::node_service::WalkDirRequest;
use protos::{
models::{PingBody, PingBodyBuilder},
node_service_time_out_client,
@@ -8,9 +11,14 @@ use protos::{
ListVolumesRequest, LocalStorageInfoRequest, MakeVolumeRequest, PingRequest, PingResponse, ReadAllRequest,
},
};
use rmp_serde::Deserializer;
use serde::Deserialize;
use rmp_serde::{Deserializer, Serializer};
use serde::{Deserialize, Serialize};
use std::{error::Error, io::Cursor};
use tokio::io::AsyncWrite;
use tokio::spawn;
use tokio::sync::mpsc;
use tonic::codegen::tokio_stream::wrappers::ReceiverStream;
use tonic::codegen::tokio_stream::StreamExt;
use tonic::Request;
const CLUSTER_ADDR: &str = "http://localhost:9000";
@@ -88,6 +96,62 @@ async fn list_volumes() -> Result<(), Box<dyn Error>> {
Ok(())
}
#[tokio::test]
async fn walk_dir() -> Result<(), Box<dyn Error>> {
println!("walk_dir");
// TODO: use writer
let opts = WalkDirOptions {
bucket: "dandan".to_owned(),
base_dir: "".to_owned(),
recursive: true,
..Default::default()
};
let (rd, mut wr) = tokio::io::duplex(1024);
let mut buf = Vec::new();
opts.serialize(&mut Serializer::new(&mut buf))?;
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
let request = Request::new(WalkDirRequest {
disk: "/home/dandan/code/rust/s3-rustfs/target/debug/data".to_string(),
walk_dir_options: buf,
});
let mut response = client.walk_dir(request).await?.into_inner();
let job1 = spawn(async move {
let mut out = MetacacheWriter::new(&mut wr);
loop {
match response.next().await {
Some(Ok(resp)) => {
if !resp.success {
println!("{}", resp.error_info.unwrap_or("".to_string()));
}
let entry = serde_json::from_str::<MetaCacheEntry>(&resp.meta_cache_entry)
.map_err(|e| ecstore::error::Error::from_string(format!("Unexpected response: {:?}", response)))
.unwrap();
out.write_obj(&entry).await.unwrap();
}
None => {
let _ = out.close().await;
break;
}
_ => {
println!("Unexpected response: {:?}", response);
let _ = out.close().await;
break;
}
}
}
});
let job2 = spawn(async move {
let mut reader = MetacacheReader::new(rd);
while let Ok(Some(entry)) = reader.peek().await {
println!("{:?}", entry);
}
});
join_all(vec![job1, job2]).await;
Ok(())
}
#[tokio::test]
async fn read_all() -> Result<(), Box<dyn Error>> {
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
ecstore/src/cache_value/metacache_set.rs
+209 -145
View File
@@ -1,3 +1,7 @@
use crate::{
disk::error::{is_err_eof, is_err_file_not_found, is_err_volume_not_found, DiskError},
metacache::writer::MetacacheReader,
};
use crate::{
disk::{DiskAPI, DiskStore, MetaCacheEntries, MetaCacheEntry, WalkDirOptions},
error::{Error, Result},
@@ -6,13 +10,9 @@ use futures::future::join_all;
use std::{future::Future, pin::Pin, sync::Arc};
use tokio::{
spawn,
sync::{
broadcast::Receiver as B_Receiver,
mpsc::{self},
RwLock,
},
sync::{broadcast::Receiver as B_Receiver, RwLock},
};
use tracing::info;
use tracing::{error, info};
type AgreedFn = Box<dyn Fn(MetaCacheEntry) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
type PartialFn = Box<dyn Fn(MetaCacheEntries, &[Option<Error>]) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + 'static>;
@@ -25,8 +25,8 @@ pub struct ListPathRawOptions {
pub bucket: String,
pub path: String,
pub recursice: bool,
pub filter_prefix: String,
pub forward_to: String,
pub filter_prefix: Option<String>,
pub forward_to: Option<String>,
pub min_disks: usize,
pub report_not_found: bool,
pub per_disk_limit: i32,
@@ -62,42 +62,41 @@ pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -
return Err(Error::from_string("list_path_raw: 0 drives provided"));
}
let mut jobs: Vec<tokio::task::JoinHandle<std::result::Result<(), Error>>> = Vec::new();
let mut readers = Vec::with_capacity(opts.disks.len());
let fds = Arc::new(RwLock::new(opts.fallback_disks.clone()));
let mut futures = Vec::with_capacity(opts.disks.len());
for disk in opts.disks.iter() {
let disk = disk.clone();
let opdisk = disk.clone();
let opts_clone = opts.clone();
let fds_clone = fds.clone();
let (m_tx, m_rx) = mpsc::channel::<MetaCacheEntry>(100);
readers.push(m_rx);
futures.push(async move {
// let (m_tx, m_rx) = mpsc::channel::<MetaCacheEntry>(100);
// readers.push(m_rx);
let (rd, mut wr) = tokio::io::duplex(64);
readers.push(MetacacheReader::new(rd));
jobs.push(spawn(async move {
let wakl_opts = WalkDirOptions {
bucket: opts_clone.bucket.clone(),
base_dir: opts_clone.path.clone(),
recursive: opts_clone.recursice,
report_notfound: opts_clone.report_not_found,
filter_prefix: opts_clone.filter_prefix.clone(),
forward_to: opts_clone.forward_to.clone(),
limit: opts_clone.per_disk_limit,
..Default::default()
};
let mut need_fallback = false;
if disk.is_none() {
need_fallback = true;
} else {
match disk
.as_ref()
.unwrap()
.walk_dir(WalkDirOptions {
bucket: opts_clone.bucket.clone(),
base_dir: opts_clone.path.clone(),
recursive: opts_clone.recursice,
report_notfound: opts_clone.report_not_found,
filter_prefix: opts_clone.filter_prefix.clone(),
forward_to: opts_clone.forward_to.clone(),
limit: opts_clone.per_disk_limit,
..Default::default()
})
.await
{
Ok(r) => {
for v in r.iter() {
let _ = m_tx.send(v.to_owned()).await;
}
if let Some(disk) = opdisk {
match disk.walk_dir(wakl_opts, &mut wr).await {
Ok(_res) => {}
Err(err) => {
error!("walk dir err {:?}", &err);
need_fallback = true;
}
Err(_) => need_fallback = true,
}
} else {
need_fallback = true;
}
while need_fallback {
@@ -106,138 +105,203 @@ pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -
if fds_w.is_empty() {
break None;
}
let fd = fds_w.remove(0);
if fd.is_some() && fd.as_ref().unwrap().is_online().await {
break fd;
if let Some(fd) = fds_w.remove(0) {
if fd.is_online().await {
break Some(fd);
}
}
};
if f_disk.is_none() {
if let Some(disk) = f_disk {
match disk
.as_ref()
.walk_dir(
WalkDirOptions {
bucket: opts_clone.bucket.clone(),
base_dir: opts_clone.path.clone(),
recursive: opts_clone.recursice,
report_notfound: opts_clone.report_not_found,
filter_prefix: opts_clone.filter_prefix.clone(),
forward_to: opts_clone.forward_to.clone(),
limit: opts_clone.per_disk_limit,
..Default::default()
},
&mut wr,
)
.await
{
Ok(_r) => {
need_fallback = false;
}
Err(err) => {
error!("walk dir2 err {:?}", &err);
break;
}
}
} else {
break;
}
match disk
.as_ref()
.unwrap()
.walk_dir(WalkDirOptions {
bucket: opts_clone.bucket.clone(),
base_dir: opts_clone.path.clone(),
recursive: opts_clone.recursice,
report_notfound: opts_clone.report_not_found,
filter_prefix: opts_clone.filter_prefix.clone(),
forward_to: opts_clone.forward_to.clone(),
limit: opts_clone.per_disk_limit,
..Default::default()
})
.await
{
Ok(r) => {
for v in r.iter() {
let _ = m_tx.send(v.to_owned()).await;
}
need_fallback = false;
}
Err(_) => break,
}
}
drop(m_tx);
});
Ok(())
}));
}
let _ = join_all(futures).await;
let revjob = spawn(async move {
let mut errs: Vec<Option<Error>> = vec![None; readers.len()];
loop {
let mut current = MetaCacheEntry::default();
let errs: Vec<Option<Error>> = vec![None; readers.len()];
loop {
let mut current = MetaCacheEntry::default();
let (mut at_eof, mut has_err, mut agree) = (0, 0, 0);
if rx.try_recv().is_ok() {
info!("list_path_raw canceled");
return Err(Error::from_string("canceled"));
}
let mut top_entries: Vec<MetaCacheEntry> = vec![MetaCacheEntry::default(); readers.len()];
// top_entries.clear();
for (i, r) in readers.iter_mut().enumerate() {
if errs[i].is_some() {
has_err += 1;
continue;
if rx.try_recv().is_ok() {
return Err(Error::from_string("canceled"));
}
let entry = match r.recv().await {
Some(entry) => entry,
None => {
at_eof += 1;
let mut top_entries: Vec<Option<MetaCacheEntry>> = vec![None; readers.len()];
let mut at_eof = 0;
let mut fnf = 0;
let mut vnf = 0;
let mut has_err = 0;
let mut agree = 0;
for (i, r) in readers.iter_mut().enumerate() {
if errs[i].is_some() {
has_err += 1;
continue;
}
};
// If no current, add it.
if current.name.is_empty() {
top_entries[i] = entry.clone();
let entry = match r.peek().await {
Ok(res) => {
if let Some(entry) = res {
entry
} else {
// eof
at_eof += 1;
continue;
}
}
Err(err) => {
if is_err_eof(&err) {
at_eof += 1;
continue;
} else if is_err_file_not_found(&err) {
at_eof += 1;
fnf += 1;
continue;
} else if is_err_volume_not_found(&err) {
at_eof += 1;
fnf += 1;
vnf += 1;
continue;
} else {
has_err += 1;
errs[i] = Some(err);
continue;
}
}
};
// If no current, add it.
if current.name.is_empty() {
top_entries.insert(i, Some(entry.clone()));
current = entry;
agree += 1;
continue;
}
// If exact match, we agree.
if let Ok((_, true)) = current.matches(&entry, true) {
top_entries.insert(i, Some(entry));
agree += 1;
continue;
}
// If only the name matches we didn't agree, but add it for resolution.
if entry.name == current.name {
top_entries.insert(i, Some(entry));
continue;
}
// We got different entries
if entry.name > current.name {
continue;
}
// We got a new, better current.
// Clear existing entries.
top_entries.clear();
agree += 1;
top_entries.insert(i, Some(entry.clone()));
current = entry;
agree += 1;
continue;
}
// If exact match, we agree.
if let Ok((_, true)) = current.matches(&entry, true) {
top_entries[i] = entry;
agree += 1;
continue;
}
// If only the name matches we didn't agree, but add it for resolution.
if entry.name == current.name {
top_entries[i] = entry;
continue;
}
// We got different entries
if entry.name > current.name {
continue;
}
// We got a new, better current.
// Clear existing entries.
for i in 0..top_entries.len() {
top_entries[i] = MetaCacheEntry::default();
}
agree += 1;
top_entries[i] = entry.clone();
current = entry;
}
if has_err > 0 && has_err > opts.disks.len() - opts.min_disks {
if let Some(finished_fn) = opts.finished.as_ref() {
finished_fn(&errs).await;
if vnf > 0 && vnf >= (readers.len() - opts.min_disks) {
return Err(Error::new(DiskError::VolumeNotFound));
}
let mut combined_err = Vec::new();
errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| match (err, disk) {
(Some(err), Some(disk)) => {
combined_err.push(format!("drive {} returned: {}", disk.to_string(), err));
}
(Some(err), None) => {
combined_err.push(err.to_string());
}
_ => {}
});
info!("list_path_raw failed, err: {:?}", combined_err);
return Err(Error::from_string(combined_err.join(", ")));
}
// Break if all at EOF or error.
if at_eof + has_err == readers.len() {
if let Some(finished_fn) = opts.finished.as_ref() {
if has_err > 0 {
if fnf > 0 && fnf >= (readers.len() - opts.min_disks) {
return Err(Error::new(DiskError::FileNotFound));
}
if has_err > 0 && has_err > opts.disks.len() - opts.min_disks {
if let Some(finished_fn) = opts.finished.as_ref() {
finished_fn(&errs).await;
}
let mut combined_err = Vec::new();
errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| match (err, disk) {
(Some(err), Some(disk)) => {
combined_err.push(format!("drive {} returned: {}", disk.to_string(), err));
}
(Some(err), None) => {
combined_err.push(err.to_string());
}
_ => {}
});
return Err(Error::from_string(combined_err.join(", ")));
}
// Break if all at EOF or error.
if at_eof + has_err == readers.len() {
if has_err > 0 {
if let Some(finished_fn) = opts.finished.as_ref() {
if has_err > 0 {
finished_fn(&errs).await;
}
}
}
break;
}
if agree == readers.len() {
for r in readers.iter_mut() {
let _ = r.skip(1).await;
}
if let Some(agreed_fn) = opts.agreed.as_ref() {
agreed_fn(current).await;
}
continue;
}
for (i, r) in readers.iter_mut().enumerate() {
if top_entries[i].is_some() {
let _ = r.skip(1).await;
}
}
if let Some(partial_fn) = opts.partial.as_ref() {
partial_fn(MetaCacheEntries(top_entries), &errs).await;
}
break;
}
Ok(())
});
if agree == readers.len() {
if let Some(agreed_fn) = opts.agreed.as_ref() {
agreed_fn(current).await;
}
continue;
}
jobs.push(revjob);
if let Some(partial_fn) = opts.partial.as_ref() {
partial_fn(MetaCacheEntries(top_entries), &errs).await;
}
}
let _ = join_all(jobs).await;
Ok(())
}
ecstore/src/config/error.rs
+3 -1
View File
@@ -1,4 +1,4 @@
use crate::{disk, error::Error};
use crate::{disk, error::Error, store_err::is_err_object_not_found};
#[derive(Debug, PartialEq, thiserror::Error)]
pub enum ConfigError {
@@ -36,6 +36,8 @@ pub fn is_not_found(err: &Error) -> bool {
ConfigError::is_not_found(e)
} else if let Some(e) = err.downcast_ref::<disk::error::DiskError>() {
matches!(e, disk::error::DiskError::FileNotFound)
} else if is_err_object_not_found(err) {
return true;
} else {
false
}
ecstore/src/disk/error.rs
+23
View File
@@ -355,6 +355,17 @@ pub fn is_err_file_not_found(err: &Error) -> bool {
matches!(err.downcast_ref::<DiskError>(), Some(DiskError::FileNotFound))
}
pub fn is_err_volume_not_found(err: &Error) -> bool {
matches!(err.downcast_ref::<DiskError>(), Some(DiskError::VolumeNotFound))
}
pub fn is_err_eof(err: &Error) -> bool {
if let Some(ioerr) = err.downcast_ref::<io::Error>() {
return ioerr.kind() == ErrorKind::UnexpectedEof;
}
false
}
pub fn is_sys_err_no_space(e: &io::Error) -> bool {
if let Some(no) = e.raw_os_error() {
return no == 28;
@@ -523,6 +534,10 @@ pub fn is_all_not_found(errs: &[Option<Error>]) -> bool {
!errs.is_empty()
}
pub fn is_all_volume_not_found(errs: &[Option<Error>]) -> bool {
DiskError::VolumeNotFound.count_errs(errs) == errs.len()
}
pub fn is_all_buckets_not_found(errs: &[Option<Error>]) -> bool {
if errs.is_empty() {
return false;
@@ -538,3 +553,11 @@ pub fn is_all_buckets_not_found(errs: &[Option<Error>]) -> bool {
}
errs.len() == not_found_count
}
pub fn is_err_os_not_exist(err: &Error) -> bool {
if let Some(os_err) = err.downcast_ref::<io::Error>() {
os_is_not_exist(os_err)
} else {
false
}
}
ecstore/src/disk/local.rs
+317 -104
View File
@@ -12,12 +12,13 @@ use crate::bitrot::bitrot_verify;
use crate::bucket::metadata_sys::{self};
use crate::cache_value::cache::{Cache, Opts, UpdateFn};
use crate::disk::error::{
convert_access_error, is_sys_err_handle_invalid, is_sys_err_invalid_arg, is_sys_err_is_dir, is_sys_err_not_dir,
map_err_not_exists, os_err_to_file_err,
convert_access_error, is_err_os_not_exist, is_sys_err_handle_invalid, is_sys_err_invalid_arg, is_sys_err_is_dir,
is_sys_err_not_dir, map_err_not_exists, os_err_to_file_err,
};
use crate::disk::os::{check_path_length, is_empty_dir};
use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
use crate::error::{Error, Result};
use crate::file_meta::read_xl_meta_no_data;
use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
use crate::heal::data_scanner::{has_active_rules, scan_data_folder, ScannerItem, ShouldSleepFn, SizeSummary};
use crate::heal::data_scanner_metric::{ScannerMetric, ScannerMetrics};
@@ -25,6 +26,7 @@ use crate::heal::data_usage_cache::{DataUsageCache, DataUsageEntry};
use crate::heal::error::{ERR_IGNORE_FILE_CONTRIB, ERR_SKIP_FILE};
use crate::heal::heal_commands::{HealScanMode, HealingTracker};
use crate::heal::heal_ops::HEALING_TRACKER_FILENAME;
use crate::metacache::writer::MetacacheWriter;
use crate::new_object_layer_fn;
use crate::set_disk::{
conv_part_err_to_int, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
@@ -33,13 +35,17 @@ use crate::set_disk::{
use crate::store_api::{BitrotAlgorithm, StorageAPI};
use crate::utils::fs::{access, lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY};
use crate::utils::os::get_info;
use crate::utils::path::{clean, has_suffix, path_join, GLOBAL_DIR_SUFFIX_WITH_SLASH, SLASH_SEPARATOR};
use crate::utils::path::{
self, clean, decode_dir_object, has_suffix, path_join, path_join_buf, GLOBAL_DIR_SUFFIX, GLOBAL_DIR_SUFFIX_WITH_SLASH,
SLASH_SEPARATOR,
};
use crate::{
file_meta::FileMeta,
store_api::{FileInfo, RawFileInfo},
utils,
};
use common::defer;
use nix::NixPath;
use path_absolutize::Absolutize;
use std::collections::{HashMap, HashSet};
use std::fmt::Debug;
@@ -55,7 +61,7 @@ use std::{
};
use time::OffsetDateTime;
use tokio::fs::{self, File};
use tokio::io::{AsyncReadExt, AsyncWriteExt, ErrorKind};
use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt, ErrorKind};
use tokio::sync::mpsc::Sender;
use tokio::sync::RwLock;
use tracing::{error, info, warn};
@@ -401,22 +407,53 @@ impl LocalDisk {
}
/// read xl.meta raw data
#[tracing::instrument(level = "debug", skip(self, volume_dir, path))]
#[tracing::instrument(level = "debug", skip(self, volume_dir, file_path))]
async fn read_raw(
&self,
bucket: &str,
volume_dir: impl AsRef<Path>,
path: impl AsRef<Path>,
file_path: impl AsRef<Path>,
read_data: bool,
) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
let meta_path = path.as_ref().join(Path::new(super::STORAGE_FORMAT_FILE));
if read_data {
self.read_all_data_with_dmtime(bucket, volume_dir, meta_path).await
} else {
self.read_all_data_with_dmtime(bucket, volume_dir, meta_path).await
// FIXME: read_metadata only suport
// self.read_metadata_with_dmtime(meta_path).await
if file_path.as_ref().is_empty() {
return Err(Error::new(DiskError::FileNotFound));
}
let meta_path = file_path.as_ref().join(Path::new(super::STORAGE_FORMAT_FILE));
let res = {
if read_data {
self.read_all_data_with_dmtime(bucket, volume_dir, meta_path).await
} else {
match self.read_metadata_with_dmtime(meta_path).await {
Ok(res) => Ok(res),
Err(err) => {
if is_err_os_not_exist(&err)
&& !skip_access_checks(volume_dir.as_ref().to_string_lossy().to_string().as_str())
{
if let Err(aerr) = access(volume_dir.as_ref()).await {
if os_is_not_exist(&aerr) {
return Err(Error::new(DiskError::VolumeNotFound));
}
}
}
if let Some(os_err) = err.downcast_ref::<std::io::Error>() {
Err(os_err_to_file_err(std::io::Error::new(os_err.kind(), os_err.to_string())))
} else {
Err(err)
}
}
}
}
};
let (buf, mtime) = res?;
if buf.is_empty() {
return Err(Error::new(DiskError::FileNotFound));
}
Ok((buf, mtime))
}
async fn read_metadata(&self, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
@@ -425,7 +462,6 @@ impl LocalDisk {
Ok(data)
}
// FIXME: read_metadata only suport
async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
@@ -444,17 +480,15 @@ impl LocalDisk {
}
let size = meta.len() as usize;
let mut bytes = Vec::new();
bytes.try_reserve_exact(size)?;
f.read_to_end(&mut bytes).await.map_err(os_err_to_file_err)?;
let data = read_xl_meta_no_data(&mut f, size).await?;
let modtime = match meta.modified() {
Ok(md) => Some(OffsetDateTime::from(md)),
Err(_) => None,
};
Ok((bytes, modtime))
Ok((data, modtime))
}
async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
@@ -500,7 +534,7 @@ impl LocalDisk {
return Err(Error::new(DiskError::UnsupportedDisk));
}
return Err(Error::new(e));
return Err(os_err_to_file_err(e));
}
};
@@ -687,6 +721,240 @@ impl LocalDisk {
let n = file.read_to_end(&mut data).await?;
bitrot_verify(&mut Cursor::new(data), n, part_size, algo, sum.to_vec(), shard_size)
}
async fn scan_dir<W: AsyncWrite + Unpin>(
&self,
current: &mut String,
opts: &WalkDirOptions,
out: &mut MetacacheWriter<W>,
objs_returned: &mut i32,
) -> Result<()> {
let forward = {
opts.forward_to.as_ref().filter(|v| v.starts_with(&*current)).map(|v| {
let forward = v.trim_start_matches(&*current);
if let Some(idx) = forward.find('/') {
forward[..idx].to_owned()
} else {
forward.to_owned()
}
})
// if let Some(forward_to) = &opts.forward_to {
// } else {
// None
// }
// if !opts.forward_to.is_empty() && opts.forward_to.starts_with(&*current) {
// let forward = opts.forward_to.trim_start_matches(&*current);
// if let Some(idx) = forward.find('/') {
// &forward[..idx]
// } else {
// forward
// }
// } else {
// ""
// }
};
if opts.limit > 0 && *objs_returned >= opts.limit {
return Ok(());
}
let mut entries = match self.list_dir("", &opts.bucket, current, -1).await {
Ok(res) => res,
Err(e) => {
if !DiskError::VolumeNotFound.is(&e) && !is_err_file_not_found(&e) {
error!("scan list_dir {}, err {:?}", &current, &e);
}
if opts.report_notfound && is_err_file_not_found(&e) && current == &opts.base_dir {
return Err(Error::new(DiskError::FileNotFound));
}
return Ok(());
}
};
if entries.is_empty() {
return Ok(());
}
let s = SLASH_SEPARATOR.chars().next().unwrap_or_default();
*current = current.trim_matches(s).to_owned();
let bucket = opts.bucket.as_str();
let mut dir_objes = HashSet::new();
// 第一层过滤
for item in entries.iter_mut() {
// warn!("walk_dir get entry {:?}", &entry);
let entry = item.clone();
// check limit
if opts.limit > 0 && *objs_returned >= opts.limit {
return Ok(());
}
// check prefix
if let Some(filter_prefix) = &opts.filter_prefix {
if !entry.starts_with(filter_prefix) {
*item = "".to_owned();
continue;
}
}
if let Some(forward) = &forward {
if &entry < forward {
*item = "".to_owned();
continue;
}
}
if entry.ends_with(SLASH_SEPARATOR) {
if entry.ends_with(GLOBAL_DIR_SUFFIX_WITH_SLASH) {
let entry = format!("{}{}", entry.as_str().trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH), SLASH_SEPARATOR);
dir_objes.insert(entry.clone());
*item = entry;
continue;
}
*item = entry.trim_end_matches(SLASH_SEPARATOR).to_owned();
continue;
}
*item = "".to_owned();
if entry.ends_with(STORAGE_FORMAT_FILE) {
//
let metadata = self
.read_metadata(self.get_object_path(bucket, format!("{}/{}", &current, &entry).as_str())?)
.await?;
let name = entry.trim_end_matches(STORAGE_FORMAT_FILE).trim_end_matches(SLASH_SEPARATOR);
let name = decode_dir_object(format!("{}/{}", &current, &name).as_str());
out.write_obj(&MetaCacheEntry {
name,
metadata,
..Default::default()
})
.await?;
*objs_returned += 1;
return Ok(());
}
}
entries.sort();
let mut entries = entries.as_slice();
if let Some(forward) = &forward {
for (i, entry) in entries.iter().enumerate() {
if entry >= forward || forward.starts_with(entry.as_str()) {
entries = &entries[i..];
break;
}
}
}
let mut dir_stack: Vec<String> = Vec::with_capacity(5);
for entry in entries.iter() {
//
if opts.limit > 0 && *objs_returned >= opts.limit {
return Ok(());
}
if entry.is_empty() {
continue;
}
let name = path::path_join_buf(&[current, entry]);
if !dir_stack.is_empty() {
if let Some(pop) = dir_stack.pop() {
if pop < name {
//
out.write_obj(&MetaCacheEntry {
name: pop.clone(),
..Default::default()
})
.await?;
if opts.recursive {
if let Err(er) = Box::pin(self.scan_dir(&mut pop.clone(), opts, out, objs_returned)).await {
error!("scan_dir err {:?}", er);
}
}
}
}
}
let mut meta = MetaCacheEntry {
name,
..Default::default()
};
let mut is_dir_obj = false;
if let Some(_dir) = dir_objes.get(entry) {
is_dir_obj = true;
meta.name
.truncate(meta.name.len() - meta.name.chars().last().unwrap().len_utf8());
meta.name.push_str(GLOBAL_DIR_SUFFIX_WITH_SLASH);
}
let fname = format!("{}/{}", &meta.name, STORAGE_FORMAT_FILE);
match self.read_metadata(self.get_object_path(&opts.bucket, fname.as_str())?).await {
Ok(res) => {
if is_dir_obj {
meta.name = meta.name.trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH).to_owned();
meta.name.push_str(SLASH_SEPARATOR);
}
meta.metadata = res;
out.write_obj(&meta).await?;
*objs_returned += 1;
}
Err(err) => {
if let Some(e) = err.downcast_ref::<std::io::Error>() {
if os_is_not_exist(e) || is_sys_err_is_dir(e) {
// NOT an object, append to stack (with slash)
// If dirObject, but no metadata (which is unexpected) we skip it.
if !is_dir_obj && !is_empty_dir(self.get_object_path(&opts.bucket, &meta.name)?).await {
meta.name.push_str(SLASH_SEPARATOR);
dir_stack.push(meta.name);
}
}
}
continue;
}
};
}
while let Some(dir) = dir_stack.pop() {
if opts.limit > 0 && *objs_returned >= opts.limit {
return Ok(());
}
out.write_obj(&MetaCacheEntry {
name: dir.clone(),
..Default::default()
})
.await?;
*objs_returned += 1;
if opts.recursive {
let mut dir = dir;
if let Err(er) = Box::pin(self.scan_dir(&mut dir, opts, out, objs_returned)).await {
warn!("scan_dir err {:?}", &er);
}
}
}
Ok(())
}
}
fn is_root_path(path: impl AsRef<Path>) -> bool {
@@ -1251,7 +1519,7 @@ impl DiskAPI for LocalDisk {
}
let volume_dir = self.get_bucket_path(volume)?;
let dir_path_abs = volume_dir.join(Path::new(&dir_path));
let dir_path_abs = volume_dir.join(Path::new(&dir_path.trim_start_matches(SLASH_SEPARATOR)));
let entries = match os::read_dir(&dir_path_abs, count).await {
Ok(res) => res,
@@ -1265,107 +1533,52 @@ impl DiskAPI for LocalDisk {
return Err(e);
}
};
Ok(entries)
}
// TODO: io.writer
async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
// warn!("walk_dir opts {:?}", &opts);
// FIXME: TODO: io.writer TODO cancel
#[tracing::instrument(level = "debug", skip(self, wr))]
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
let volume_dir = self.get_bucket_path(&opts.bucket)?;
let mut metas = Vec::new();
if !skip_access_checks(&opts.bucket) {
if let Err(e) = access(&volume_dir).await {
return Err(convert_access_error(e, DiskError::VolumeAccessDenied));
}
}
let mut wr = wr;
let mut out = MetacacheWriter::new(&mut wr);
let mut objs_returned = 0;
if opts.base_dir.ends_with(SLASH_SEPARATOR) {
let fpath = self.get_object_path(
&opts.bucket,
format!("{}/{}", opts.base_dir.trim_end_matches(SLASH_SEPARATOR), STORAGE_FORMAT_FILE).as_str(),
path_join_buf(&[
format!("{}{}", opts.base_dir.trim_end_matches(SLASH_SEPARATOR), GLOBAL_DIR_SUFFIX).as_str(),
STORAGE_FORMAT_FILE,
])
.as_str(),
)?;
if let Ok(data) = self.read_metadata(fpath).await {
let meta = MetaCacheEntry {
name: opts.base_dir.clone(),
metadata: data,
..Default::default()
};
metas.push(meta);
return Ok(metas);
out.write_obj(&meta).await?;
objs_returned += 1;
}
}
let mut entries = match self.list_dir("", &opts.bucket, &opts.base_dir, -1).await {
Ok(res) => res,
Err(e) => {
if !DiskError::VolumeNotFound.is(&e) && !is_err_file_not_found(&e) {
info!("list_dir err {:?}", &e);
}
let mut current = opts.base_dir.clone();
self.scan_dir(&mut current, &opts, &mut out, &mut objs_returned).await?;
if opts.report_notfound && is_err_file_not_found(&e) {
return Err(e);
}
return Ok(Vec::new());
}
};
if entries.is_empty() {
return Ok(Vec::new());
}
entries.sort();
// 已读计数
let objs_returned = 0;
let bucket = opts.bucket.as_str();
let mut dir_objes = HashSet::new();
// 第一层过滤
for entry in entries.iter() {
// warn!("walk_dir get entry {:?}", &entry);
// check limit
if opts.limit > 0 && objs_returned >= opts.limit {
return Ok(metas);
}
// check prefix
if !opts.filter_prefix.is_empty() && !entry.starts_with(&opts.filter_prefix) {
continue;
}
let mut meta = MetaCacheEntry { ..Default::default() };
let mut name = {
if opts.base_dir.is_empty() {
entry.clone()
} else {
format!("{}{}{}", opts.base_dir.trim_end_matches(SLASH_SEPARATOR), SLASH_SEPARATOR, entry)
}
};
if name.ends_with(SLASH_SEPARATOR) {
if name.ends_with(GLOBAL_DIR_SUFFIX_WITH_SLASH) {
name = format!("{}{}", name.as_str().trim_end_matches(GLOBAL_DIR_SUFFIX_WITH_SLASH), SLASH_SEPARATOR);
dir_objes.insert(name.clone());
} else {
name = name.as_str().trim_end_matches(SLASH_SEPARATOR).to_owned();
}
}
meta.name = name;
let fpath = self.get_object_path(bucket, format!("{}/{}", &meta.name, STORAGE_FORMAT_FILE).as_str())?;
if let Ok(data) = self.read_metadata(&fpath).await {
meta.metadata = data;
} else {
let fpath = self.get_object_path(bucket, &meta.name)?;
if !is_empty_dir(fpath).await {
meta.name = format!("{}{}", &meta.name, SLASH_SEPARATOR);
}
}
metas.push(meta);
}
Ok(metas)
Ok(())
}
// #[tracing::instrument(skip(self))]
@@ -1447,7 +1660,7 @@ impl DiskAPI for LocalDisk {
let mut xlmeta = FileMeta::new();
if let Some(dst_buf) = has_dst_buf.as_ref() {
if FileMeta::is_xl_format(dst_buf) {
if FileMeta::is_xl2_v1_format(dst_buf) {
if let Ok(nmeta) = FileMeta::load(dst_buf) {
xlmeta = nmeta
}
@@ -1695,7 +1908,7 @@ impl DiskAPI for LocalDisk {
}
})?;
if !FileMeta::is_xl_format(buf.as_slice()) {
if !FileMeta::is_xl2_v1_format(buf.as_slice()) {
return Err(Error::new(DiskError::FileVersionNotFound));
}
ecstore/src/disk/mod.rs
+308 -24
View File
@@ -16,17 +16,20 @@ pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp";
use crate::utils::proto_err_to_err;
use crate::{
bucket::{metadata_sys::get_versioning_config, versioning::VersioningApi},
erasure::Writer,
error::{Error, Result},
file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion},
file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion, VersionType},
heal::{
data_scanner::ShouldSleepFn,
data_usage_cache::{DataUsageCache, DataUsageEntry},
heal_commands::{HealScanMode, HealingTracker},
},
store_api::{FileInfo, RawFileInfo},
store_api::{FileInfo, ObjectInfo, RawFileInfo},
utils::path::SLASH_SEPARATOR,
};
use endpoint::Endpoint;
use error::DiskError;
use futures::StreamExt;
use local::LocalDisk;
use madmin::info_commands::DiskMetrics;
@@ -46,7 +49,7 @@ use std::{
use time::OffsetDateTime;
use tokio::{
fs::File,
io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt},
io::{AsyncReadExt, AsyncSeekExt, AsyncWrite, AsyncWriteExt},
sync::mpsc::{self, Sender},
};
use tokio_stream::wrappers::ReceiverStream;
@@ -210,10 +213,10 @@ impl DiskAPI for Disk {
}
}
async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.walk_dir(opts).await,
Disk::Remote(remote_disk) => remote_disk.walk_dir(opts).await,
Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await,
Disk::Remote(remote_disk) => remote_disk.walk_dir(opts, wr).await,
}
}
@@ -405,8 +408,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn stat_volume(&self, volume: &str) -> Result<VolumeInfo>;
async fn delete_volume(&self, volume: &str) -> Result<()>;
// 并发边读边写 TODO: wr io.Writer
async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>>;
// 并发边读边写 w <- MetaCacheEntry
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()>;
// Metadata operations
async fn delete_version(
@@ -557,6 +560,18 @@ pub struct FileInfoVersions {
pub free_versions: Vec<FileInfo>,
}
impl FileInfoVersions {
pub fn find_version_index(&self, v: &str) -> Option<usize> {
if v.is_empty() {
return None;
}
let vid = Uuid::parse_str(v).unwrap_or(Uuid::nil());
self.versions.iter().position(|v| v.version_id == Some(vid))
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct WalkDirOptions {
// Bucket to scanner
@@ -571,10 +586,10 @@ pub struct WalkDirOptions {
// FilterPrefix will only return results with given prefix within folder.
// Should never contain a slash.
pub filter_prefix: String,
pub filter_prefix: Option<String>,
// ForwardTo will forward to the given object path.
pub forward_to: String,
pub forward_to: Option<String>,
// Limit the number of returned objects if > 0.
pub limit: i32,
@@ -594,7 +609,7 @@ pub struct MetadataResolutionParams {
pub candidates: Vec<Vec<FileMetaShallowVersion>>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)]
pub struct MetaCacheEntry {
// name is the full name of the object including prefixes
pub name: String,
@@ -603,10 +618,10 @@ pub struct MetaCacheEntry {
pub metadata: Vec<u8>,
// cached contains the metadata if decoded.
cached: Option<FileMeta>,
pub cached: Option<FileMeta>,
// Indicates the entry can be reused and only one reference to metadata is expected.
_reusable: bool,
pub reusable: bool,
}
impl MetaCacheEntry {
@@ -620,39 +635,92 @@ impl MetaCacheEntry {
Ok(wr)
}
pub fn is_dir(&self) -> bool {
self.metadata.is_empty() && self.name.ends_with('/')
}
pub fn is_in_dir(&self, dir: &str, separator: &str) -> bool {
if dir.is_empty() {
let idx = self.name.find(separator);
return idx.is_none() || idx.unwrap() == self.name.len() - separator.len();
}
let ext = self.name.trim_start_matches(dir);
if ext.len() != self.name.len() {
let idx = ext.find(separator);
return idx.is_none() || idx.unwrap() == ext.len() - separator.len();
}
false
}
pub fn is_object(&self) -> bool {
!self.metadata.is_empty()
}
pub fn is_object_dir(&self) -> bool {
!self.metadata.is_empty() && self.name.ends_with(SLASH_SEPARATOR)
}
pub fn is_latest_deletemarker(&mut self) -> bool {
if let Some(cached) = &self.cached {
if cached.versions.is_empty() {
return true;
}
return cached.versions[0].header.version_type == VersionType::Delete;
}
if !FileMeta::is_xl2_v1_format(&self.metadata) {
return false;
}
match FileMeta::check_xl2_v1(&self.metadata) {
Ok((meta, _, _)) => {
if !meta.is_empty() {
// TODO: IsLatestDeleteMarker
}
}
Err(_) => return true,
}
match self.xl_meta() {
Ok(res) => {
if res.versions.is_empty() {
return true;
}
res.versions[0].header.version_type == VersionType::Delete
}
Err(_) => true,
}
}
#[tracing::instrument(level = "debug", skip(self))]
pub fn to_fileinfo(&self, bucket: &str) -> Result<Option<FileInfo>> {
pub fn to_fileinfo(&self, bucket: &str) -> Result<FileInfo> {
if self.is_dir() {
return Ok(Some(FileInfo {
return Ok(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 {
return Ok(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));
return Ok(fi);
}
let mut fm = FileMeta::new();
@@ -660,7 +728,7 @@ impl MetaCacheEntry {
let fi = fm.into_fileinfo(bucket, self.name.as_str(), "", false, false)?;
return Ok(Some(fi));
return Ok(fi);
}
pub fn file_info_versions(&self, bucket: &str) -> Result<FileInfoVersions> {
@@ -767,11 +835,36 @@ impl MetaCacheEntry {
Ok((prefer, true))
}
pub fn xl_meta(&mut self) -> Result<FileMeta> {
if self.is_dir() {
return Err(Error::new(DiskError::FileNotFound));
}
if let Some(meta) = &self.cached {
Ok(meta.clone())
} else {
if self.metadata.is_empty() {
return Err(Error::new(DiskError::FileNotFound));
}
let meta = FileMeta::load(&self.metadata)?;
self.cached = Some(meta.clone());
Ok(meta)
}
}
}
pub struct MetaCacheEntries(pub Vec<MetaCacheEntry>);
#[derive(Debug, Default)]
pub struct MetaCacheEntries(pub Vec<Option<MetaCacheEntry>>);
impl MetaCacheEntries {
#[allow(clippy::should_implement_trait)]
pub fn as_ref(&self) -> &[Option<MetaCacheEntry>] {
&self.0
}
pub fn resolve(&self, mut params: MetadataResolutionParams) -> Result<Option<MetaCacheEntry>> {
if self.0.is_empty() {
return Ok(None);
@@ -784,7 +877,7 @@ impl MetaCacheEntries {
let mut objs_agree = 0;
let mut objs_valid = 0;
for entry in self.0.iter() {
for entry in self.0.iter().flatten() {
if entry.name.is_empty() {
continue;
}
@@ -842,7 +935,7 @@ impl MetaCacheEntries {
meta_ver: selected.as_ref().unwrap().cached.as_ref().unwrap().meta_ver,
..Default::default()
}),
_reusable: true,
reusable: true,
..Default::default()
});
@@ -858,7 +951,198 @@ impl MetaCacheEntries {
}
pub fn first_found(&self) -> (Option<MetaCacheEntry>, usize) {
(self.0.iter().find(|x| !x.name.is_empty()).cloned(), self.0.len())
(self.0.iter().find(|x| x.is_some()).cloned().unwrap_or_default(), self.0.len())
}
}
#[derive(Debug, Default)]
pub struct MetaCacheEntriesSortedResult {
pub entries: Option<MetaCacheEntriesSorted>,
pub err: Option<Error>,
}
// impl MetaCacheEntriesSortedResult {
// pub fn entriy_list(&self) -> Vec<&MetaCacheEntry> {
// if let Some(entries) = &self.entries {
// entries.entries()
// } else {
// Vec::new()
// }
// }
// }
#[derive(Debug, Default)]
pub struct MetaCacheEntriesSorted {
pub o: MetaCacheEntries,
pub list_id: Option<String>,
pub reuse: bool,
pub last_skipped_entry: Option<String>,
}
impl MetaCacheEntriesSorted {
pub fn entries(&self) -> Vec<&MetaCacheEntry> {
let entries: Vec<&MetaCacheEntry> = self.o.0.iter().flatten().collect();
entries
}
pub fn forward_past(&mut self, marker: Option<String>) {
if let Some(val) = marker {
// TODO: reuse
if let Some(idx) = self.o.0.iter().flatten().position(|v| v.name > val) {
self.o.0 = self.o.0.split_off(idx);
}
}
}
pub async fn file_infos(&self, bucket: &str, prefix: &str, delimiter: Option<String>) -> Vec<ObjectInfo> {
let vcfg = get_versioning_config(bucket).await.ok();
let mut objects = Vec::with_capacity(self.o.as_ref().len());
let mut prev_prefix = "";
for entry in self.o.as_ref().iter().flatten() {
if entry.is_object() {
if let Some(delimiter) = &delimiter {
if let Some(idx) = entry.name.trim_start_matches(prefix).find(delimiter) {
let idx = prefix.len() + idx + delimiter.len();
if let Some(curr_prefix) = entry.name.get(0..idx) {
if curr_prefix == prev_prefix {
continue;
}
prev_prefix = curr_prefix;
objects.push(ObjectInfo {
is_dir: true,
bucket: bucket.to_owned(),
name: curr_prefix.to_owned(),
..Default::default()
});
}
continue;
}
}
if let Ok(fi) = entry.to_fileinfo(bucket) {
// TODO:VersionPurgeStatus
let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
objects.push(fi.to_object_info(bucket, &entry.name, versioned));
}
continue;
}
if entry.is_dir() {
if let Some(delimiter) = &delimiter {
if let Some(idx) = entry.name.trim_start_matches(prefix).find(delimiter) {
let idx = prefix.len() + idx + delimiter.len();
if let Some(curr_prefix) = entry.name.get(0..idx) {
if curr_prefix == prev_prefix {
continue;
}
prev_prefix = curr_prefix;
objects.push(ObjectInfo {
is_dir: true,
bucket: bucket.to_owned(),
name: curr_prefix.to_owned(),
..Default::default()
});
}
}
}
}
}
objects
}
pub async fn file_info_versions(
&self,
bucket: &str,
prefix: &str,
delimiter: Option<String>,
after_v: Option<String>,
) -> Vec<ObjectInfo> {
let vcfg = get_versioning_config(bucket).await.ok();
let mut objects = Vec::with_capacity(self.o.as_ref().len());
let mut prev_prefix = "";
let mut after_v = after_v;
for entry in self.o.as_ref().iter().flatten() {
if entry.is_object() {
if let Some(delimiter) = &delimiter {
if let Some(idx) = entry.name.trim_start_matches(prefix).find(delimiter) {
let idx = prefix.len() + idx + delimiter.len();
if let Some(curr_prefix) = entry.name.get(0..idx) {
if curr_prefix == prev_prefix {
continue;
}
prev_prefix = curr_prefix;
objects.push(ObjectInfo {
is_dir: true,
bucket: bucket.to_owned(),
name: curr_prefix.to_owned(),
..Default::default()
});
}
continue;
}
}
let mut fiv = match entry.file_info_versions(bucket) {
Ok(res) => res,
Err(_err) => {
//
continue;
}
};
let fi_versions = 'c: {
if let Some(after_val) = &after_v {
if let Some(idx) = fiv.find_version_index(after_val) {
after_v = None;
break 'c fiv.versions.split_off(idx + 1);
}
after_v = None;
break 'c fiv.versions;
} else {
break 'c fiv.versions;
}
};
for fi in fi_versions.into_iter() {
// VersionPurgeStatus
let versioned = vcfg.clone().map(|v| v.0.versioned(&entry.name)).unwrap_or_default();
objects.push(fi.to_object_info(bucket, &entry.name, versioned));
}
continue;
}
if entry.is_dir() {
if let Some(delimiter) = &delimiter {
if let Some(idx) = entry.name.trim_start_matches(prefix).find(delimiter) {
let idx = prefix.len() + idx + delimiter.len();
if let Some(curr_prefix) = entry.name.get(0..idx) {
if curr_prefix == prev_prefix {
continue;
}
prev_prefix = curr_prefix;
objects.push(ObjectInfo {
is_dir: true,
bucket: bucket.to_owned(),
name: curr_prefix.to_owned(),
..Default::default()
});
}
}
}
}
}
objects
}
}
ecstore/src/disk/remote.rs
+31 -21
View File
@@ -10,7 +10,12 @@ use protos::{
StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
},
};
use tokio::sync::mpsc::{self, Sender};
use rmp_serde::Serializer;
use serde::Serialize;
use tokio::{
io::AsyncWrite,
sync::mpsc::{self, Sender},
};
use tokio_stream::{wrappers::ReceiverStream, StreamExt};
use tonic::Request;
use tracing::info;
@@ -18,8 +23,8 @@ use uuid::Uuid;
use super::{
endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption,
FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader,
RemoteFileWriter, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
FileInfoVersions, FileReader, FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter,
RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
};
use crate::utils::proto_err_to_err;
use crate::{
@@ -32,6 +37,7 @@ use crate::{
},
store_api::{FileInfo, RawFileInfo},
};
use crate::{disk::MetaCacheEntry, metacache::writer::MetacacheWriter};
use protos::proto_gen::node_service::RenamePartRequst;
#[derive(Debug)]
@@ -376,34 +382,38 @@ impl DiskAPI for RemoteDisk {
Ok(response.volumes)
}
async fn walk_dir(&self, opts: WalkDirOptions) -> Result<Vec<MetaCacheEntry>> {
// FIXME: TODO: use writer
async fn walk_dir<W: AsyncWrite + Unpin + Send>(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> {
info!("walk_dir");
let walk_dir_options = serde_json::to_string(&opts)?;
let mut wr = wr;
let mut out = MetacacheWriter::new(&mut wr);
let mut buf = Vec::new();
opts.serialize(&mut Serializer::new(&mut buf))?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| Error::from_string(format!("can not get client, err: {}", err)))?;
let request = Request::new(WalkDirRequest {
disk: self.endpoint.to_string(),
walk_dir_options,
walk_dir_options: buf,
});
let mut response = client.walk_dir(request).await?.into_inner();
let response = client.walk_dir(request).await?.into_inner();
if !response.success {
return if let Some(err) = &response.error {
Err(proto_err_to_err(err))
} else {
Err(Error::from_string(""))
};
loop {
match response.next().await {
Some(Ok(resp)) => {
if !resp.success {
return Err(Error::from_string(resp.error_info.unwrap_or("".to_string())));
}
let entry = serde_json::from_str::<MetaCacheEntry>(&resp.meta_cache_entry)
.map_err(|_| Error::from_string(format!("Unexpected response: {:?}", response)))?;
out.write_obj(&entry).await?;
}
None => break,
_ => return Err(Error::from_string(format!("Unexpected response: {:?}", response))),
}
}
let entries = response
.meta_cache_entry
.into_iter()
.filter_map(|json_str| serde_json::from_str::<MetaCacheEntry>(&json_str).ok())
.collect();
Ok(entries)
Ok(())
}
async fn rename_data(
ecstore/src/endpoints.rs
+1 -2
View File
@@ -1,5 +1,4 @@
use tracing::{info, warn};
use url::Url;
use tracing::warn;
use crate::{
disk::endpoint::{Endpoint, EndpointType},
ecstore/src/error.rs
+2 -5
View File
@@ -1,9 +1,6 @@
use std::io;
use tracing::warn;
use tracing_error::{SpanTrace, SpanTraceStatus};
use crate::disk::error::{clone_disk_err, DiskError};
use std::io;
use tracing_error::{SpanTrace, SpanTraceStatus};
pub type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
ecstore/src/file_meta.rs
+152 -16
View File
@@ -3,10 +3,11 @@ use rmp::Marker;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::fmt::Display;
use std::io::{Read, Write};
use std::io::{self, Read, Write};
use std::{collections::HashMap, io::Cursor};
use time::OffsetDateTime;
use tracing::warn;
use tokio::io::AsyncRead;
use tracing::{error, warn};
use uuid::Uuid;
use xxhash_rust::xxh64;
@@ -35,6 +36,9 @@ const XL_FLAG_FREE_VERSION: u8 = 1 << 0;
// const XL_FLAG_USES_DATA_DIR: u8 = 1 << 1;
const _XL_FLAG_INLINE_DATA: u8 = 1 << 2;
const META_DATA_READ_DEFAULT: usize = 4 << 10;
const MSGP_UINT32_SIZE: usize = 5;
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct FileMeta {
pub versions: Vec<FileMetaShallowVersion>,
@@ -51,8 +55,9 @@ impl FileMeta {
}
}
pub fn is_xl_format(buf: &[u8]) -> bool {
!matches!(Self::read_xl_file_header(buf), Err(_e))
// isXL2V1Format
pub fn is_xl2_v1_format(buf: &[u8]) -> bool {
!matches!(Self::check_xl2_v1(buf), Err(_e))
}
pub fn load(buf: &[u8]) -> Result<FileMeta> {
@@ -62,9 +67,10 @@ impl FileMeta {
Ok(xl)
}
// read_xl_file_header 读xl文件头，返回后续内容，版本信息
// check_xl2_v1 读xl文件头，返回后续内容，版本信息
// checkXL2V1
#[tracing::instrument]
pub fn read_xl_file_header(buf: &[u8]) -> Result<(&[u8], u16, u16)> {
pub fn check_xl2_v1(buf: &[u8]) -> Result<(&[u8], u16, u16)> {
if buf.len() < 8 {
return Err(Error::msg("xl file header not exists"));
}
@@ -81,12 +87,22 @@ impl FileMeta {
Ok((&buf[8..], major, minor))
}
// 固定u32
pub fn read_bytes_header(buf: &[u8]) -> Result<(u32, &[u8])> {
let (mut size_buf, _) = buf.split_at(5);
// 取meta数据，buf = crc + data
let bin_len = rmp::decode::read_bin_len(&mut size_buf)?;
Ok((bin_len, &buf[5..]))
}
#[tracing::instrument]
pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result<u64> {
let i = buf.len() as u64;
// check version, buf = buf[8..]
let (buf, _, _) = Self::read_xl_file_header(buf)?;
let (buf, _, _) = Self::check_xl2_v1(buf)?;
let (mut size_buf, buf) = buf.split_at(5);
@@ -484,7 +500,7 @@ impl FileMeta {
}
}
let mut fi = ver.into_fileinfo(volume, path, has_vid, all_parts)?;
let mut fi = ver.to_fileinfo(volume, path, has_vid, all_parts)?;
fi.is_latest = is_latest;
if let Some(_d) = succ_mod_time {
fi.successor_mod_time = succ_mod_time;
@@ -511,7 +527,7 @@ impl FileMeta {
for version in self.versions.iter() {
let mut file_version = FileMetaVersion::default();
file_version.unmarshal_msg(&version.meta)?;
let fi = file_version.into_fileinfo(volume, path, None, all_parts);
let fi = file_version.to_fileinfo(volume, path, None, all_parts);
versions.push(fi);
}
@@ -553,10 +569,10 @@ pub struct FileMetaShallowVersion {
}
impl FileMetaShallowVersion {
pub fn into_fileinfo(&self, volume: &str, path: &str, version_id: Option<Uuid>, all_parts: bool) -> Result<FileInfo> {
pub fn to_fileinfo(&self, volume: &str, path: &str, version_id: Option<Uuid>, all_parts: bool) -> Result<FileInfo> {
let file_version = FileMetaVersion::try_from(self.meta.as_slice())?;
Ok(file_version.into_fileinfo(volume, path, version_id, all_parts))
Ok(file_version.to_fileinfo(volume, path, version_id, all_parts))
}
}
@@ -760,7 +776,7 @@ impl FileMetaVersion {
FileMetaVersionHeader::from(self.clone())
}
pub fn into_fileinfo(self, volume: &str, path: &str, version_id: Option<Uuid>, all_parts: bool) -> FileInfo {
pub fn to_fileinfo(&self, volume: &str, path: &str, version_id: Option<Uuid>, all_parts: bool) -> FileInfo {
match self.version_type {
VersionType::Invalid => FileInfo {
name: path.to_string(),
@@ -1994,6 +2010,89 @@ async fn get_file_info(buf: &[u8], volume: &str, path: &str, version_id: &str, o
Ok(fi)
}
async fn read_more<R: AsyncRead + Unpin>(
reader: &mut R,
buf: &mut Vec<u8>,
total_size: usize,
read_size: usize,
has_full: bool,
) -> Result<()> {
use tokio::io::AsyncReadExt;
let has = buf.len();
if has >= read_size {
return Ok(());
}
if has_full || read_size > total_size {
return Err(Error::new(io::Error::new(io::ErrorKind::UnexpectedEof, "Unexpected EOF")));
}
let extra = read_size - has;
if buf.capacity() >= read_size {
// Extend the buffer if we have enough space.
buf.resize(read_size, 0);
} else {
buf.extend(vec![0u8; extra]);
}
reader.read_exact(&mut buf[has..]).await?;
Ok(())
}
pub async fn read_xl_meta_no_data<R: AsyncRead + Unpin>(reader: &mut R, size: usize) -> Result<Vec<u8>> {
use tokio::io::AsyncReadExt;
let mut initial = size;
let mut has_full = true;
if initial > META_DATA_READ_DEFAULT {
initial = META_DATA_READ_DEFAULT;
has_full = false;
}
let mut buf = vec![0u8; initial];
reader.read_exact(&mut buf).await?;
let (tmp_buf, major, minor) = FileMeta::check_xl2_v1(&buf)?;
match major {
1 => match minor {
0 => {
read_more(reader, &mut buf, size, size, has_full).await?;
Ok(buf)
}
1..=3 => {
let (sz, tmp_buf) = FileMeta::read_bytes_header(tmp_buf)?;
let mut want = sz as usize + (buf.len() - tmp_buf.len());
if minor < 2 {
read_more(reader, &mut buf, size, want, has_full).await?;
return Ok(buf[..want].to_vec());
}
let want_max = usize::min(want + MSGP_UINT32_SIZE, size);
read_more(reader, &mut buf, size, want_max, has_full).await?;
if buf.len() < want {
error!("read_xl_meta_no_data buffer too small (length: {}, needed: {})", &buf.len(), want);
return Err(Error::new(DiskError::FileCorrupt));
}
let tmp = &buf[want..];
let crc_size = 5;
let other_size = tmp.len() - crc_size;
want += tmp.len() - other_size;
Ok(buf[..want].to_vec())
}
_ => Err(Error::new(io::Error::new(io::ErrorKind::InvalidData, "Unknown minor metadata version"))),
},
_ => Err(Error::new(io::Error::new(io::ErrorKind::InvalidData, "Unknown major metadata version"))),
}
}
#[cfg(test)]
mod test {
@@ -2012,15 +2111,11 @@ mod test {
fm.add_version(fi).unwrap();
}
// println!("fm:{:?}", &fm);
let buff = fm.marshal_msg().unwrap();
let mut newfm = FileMeta::default();
newfm.unmarshal_msg(&buff).unwrap();
// println!("newone:{:?}", newone);
assert_eq!(fm, newfm)
}
@@ -2107,3 +2202,44 @@ mod test {
assert_eq!(obj.version_id, vid);
}
}
#[tokio::test]
async fn test_read_xl_meta_no_data() {
use tokio::fs;
use tokio::fs::File;
use tokio::io::AsyncWriteExt;
let mut fm = FileMeta::new();
let (m, n) = (3, 2);
for i in 0..5 {
let mut fi = FileInfo::new(i.to_string().as_str(), m, n);
fi.mod_time = Some(OffsetDateTime::now_utc());
fm.add_version(fi).unwrap();
}
let mut buff = fm.marshal_msg().unwrap();
buff.resize(buff.len() + 100, 0);
let filepath = "./test_xl.meta";
let mut file = File::create(filepath).await.unwrap();
// 写入字符串
file.write_all(&buff).await.unwrap();
let mut f = File::open(filepath).await.unwrap();
let stat = f.metadata().await.unwrap();
let data = read_xl_meta_no_data(&mut f, stat.len() as usize).await.unwrap();
let mut newfm = FileMeta::default();
newfm.unmarshal_msg(&data).unwrap();
fs::remove_file(filepath).await.unwrap();
assert_eq!(fm, newfm)
}
ecstore/src/io.rs
+226
View File
@@ -0,0 +1,226 @@
use std::io::Read;
use std::io::Write;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::fs::File;
use tokio::io::{self, AsyncRead, AsyncWrite, ReadBuf};
pub enum Reader {
File(File),
Buffer(VecAsyncReader),
}
impl AsyncRead for Reader {
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
match self.get_mut() {
Reader::File(file) => Pin::new(file).poll_read(cx, buf),
Reader::Buffer(buffer) => Pin::new(buffer).poll_read(cx, buf),
}
}
}
#[derive(Default)]
pub enum Writer {
#[default]
NotUse,
File(File),
Buffer(VecAsyncWriter),
}
impl AsyncWrite for Writer {
fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
match self.get_mut() {
Writer::File(file) => Pin::new(file).poll_write(cx, buf),
Writer::Buffer(buff) => Pin::new(buff).poll_write(cx, buf),
Writer::NotUse => Poll::Ready(Ok(0)),
}
}
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
match self.get_mut() {
Writer::File(file) => Pin::new(file).poll_flush(cx),
Writer::Buffer(buff) => Pin::new(buff).poll_flush(cx),
Writer::NotUse => Poll::Ready(Ok(())),
}
}
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
match self.get_mut() {
Writer::File(file) => Pin::new(file).poll_shutdown(cx),
Writer::Buffer(buff) => Pin::new(buff).poll_shutdown(cx),
Writer::NotUse => Poll::Ready(Ok(())),
}
}
}
pub struct AsyncToSync<R> {
inner: R,
}
impl<R: AsyncRead + Unpin> AsyncToSync<R> {
pub fn new_reader(inner: R) -> Self {
Self { inner }
}
fn read_async(&mut self, cx: &mut Context<'_>, buf: &mut [u8]) -> Poll<std::io::Result<usize>> {
let mut read_buf = ReadBuf::new(buf);
// Poll the underlying AsyncRead to fill the ReadBuf
match Pin::new(&mut self.inner).poll_read(cx, &mut read_buf) {
Poll::Ready(Ok(())) => Poll::Ready(Ok(read_buf.filled().len())),
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
Poll::Pending => Poll::Pending,
}
}
}
impl<R: AsyncWrite + Unpin> AsyncToSync<R> {
pub fn new_writer(inner: R) -> Self {
Self { inner }
}
// This function will perform a write using AsyncWrite
fn write_async(&mut self, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
let result = Pin::new(&mut self.inner).poll_write(cx, buf);
match result {
Poll::Ready(Ok(n)) => Poll::Ready(Ok(n)),
Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
Poll::Pending => Poll::Pending,
}
}
// This function will perform a flush using AsyncWrite
fn flush_async(&mut self, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
Pin::new(&mut self.inner).poll_flush(cx)
}
}
impl<R: AsyncRead + Unpin> Read for AsyncToSync<R> {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
loop {
match self.read_async(&mut cx, buf) {
Poll::Ready(Ok(n)) => return Ok(n),
Poll::Ready(Err(e)) => return Err(e),
Poll::Pending => {
// If Pending, we need to wait for the readiness.
// Here, we can use an arbitrary mechanism to yield control,
// this might be blocking until some readiness occurs can be complex.
// A full blocking implementation would require an async runtime to block on.
std::thread::sleep(std::time::Duration::from_millis(1)); // Replace with proper waiting if needed
}
}
}
}
}
impl<W: AsyncWrite + Unpin> Write for AsyncToSync<W> {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
loop {
match self.write_async(&mut cx, buf) {
Poll::Ready(Ok(n)) => return Ok(n),
Poll::Ready(Err(e)) => return Err(e),
Poll::Pending => {
// Here we are blocking and waiting for the async operation to complete.
std::thread::sleep(std::time::Duration::from_millis(1)); // Not efficient, see notes.
}
}
}
}
fn flush(&mut self) -> std::io::Result<()> {
let mut cx = std::task::Context::from_waker(futures::task::noop_waker_ref());
loop {
match self.flush_async(&mut cx) {
Poll::Ready(Ok(())) => return Ok(()),
Poll::Ready(Err(e)) => return Err(e),
Poll::Pending => {
// Again, blocking to wait for flush.
std::thread::sleep(std::time::Duration::from_millis(1)); // Not efficient, see notes.
}
}
}
}
}
pub struct VecAsyncWriter {
buffer: Vec<u8>,
}
impl VecAsyncWriter {
/// Create a new VecAsyncWriter with an empty Vec<u8>.
pub fn new(buffer: Vec<u8>) -> Self {
VecAsyncWriter { buffer }
}
/// Retrieve the underlying buffer.
pub fn get_buffer(&self) -> &[u8] {
&self.buffer
}
}
// Implementing AsyncWrite trait for VecAsyncWriter
impl AsyncWrite for VecAsyncWriter {
fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
let len = buf.len();
// Assume synchronous writing for simplicity
self.get_mut().buffer.extend_from_slice(buf);
// Returning the length of written data
Poll::Ready(Ok(len))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
// In this case, flushing is a no-op for a Vec<u8>
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
// Similar to flush, shutdown has no effect here
Poll::Ready(Ok(()))
}
}
pub struct VecAsyncReader {
buffer: Vec<u8>,
position: usize,
}
impl VecAsyncReader {
/// Create a new VecAsyncReader with the given Vec<u8>.
pub fn new(buffer: Vec<u8>) -> Self {
VecAsyncReader { buffer, position: 0 }
}
/// Reset the reader position.
pub fn reset(&mut self) {
self.position = 0;
}
}
// Implementing AsyncRead trait for VecAsyncReader
impl AsyncRead for VecAsyncReader {
fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf) -> Poll<io::Result<()>> {
let this = self.get_mut();
// Check how many bytes are available to read
let len = this.buffer.len();
let bytes_available = len - this.position;
if bytes_available == 0 {
// If there's no more data to read, return ready with an Eof
return Poll::Ready(Ok(()));
}
// Calculate how much we can read into the provided buffer
let to_read = std::cmp::min(bytes_available, buf.remaining());
// Write the data to the buf
buf.put_slice(&this.buffer[this.position..this.position + to_read]);
// Update the position
this.position += to_read;
// Indicate how many bytes were read
Poll::Ready(Ok(()))
}
}
ecstore/src/lib.rs
+8 -7
View File
@@ -1,5 +1,6 @@
pub mod admin_server_info;
pub mod bitrot;
pub mod bucket;
pub mod cache_value;
mod chunk_stream;
pub mod config;
@@ -9,25 +10,25 @@ pub mod endpoints;
pub mod erasure;
pub mod error;
mod file_meta;
pub mod file_meta_inline;
pub mod global;
pub mod heal;
pub mod io;
pub mod metacache;
pub mod metrics_realtime;
pub mod notification_sys;
pub mod peer;
pub mod peer_rest_client;
pub mod pools;
mod quorum;
pub mod set_disk;
mod sets;
pub mod store;
pub mod store_api;
mod store_init;
pub mod utils;
pub mod bucket;
pub mod file_meta_inline;
pub mod pools;
pub mod store_err;
mod store_init;
pub mod store_list_objects;
pub mod utils;
pub mod xhttp;
pub use global::new_object_layer_fn;
ecstore/src/list_objects.rs
-290
View File
@@ -1,290 +0,0 @@
use crate::disk::WalkDirOptions;
use crate::error::{Error, Result};
use crate::peer::is_reserved_or_invalid_bucket;
use crate::store::check_list_objs_args;
use crate::store_api::{ListObjectsInfo, ObjectInfo};
use crate::utils::path::{base_dir_from_prefix, SLASH_SEPARATOR};
use crate::{store::ECStore, store_api::ListObjectsV2Info};
use futures::future::join_all;
use std::collections::HashSet;
use std::io::ErrorKind;
const MAX_OBJECT_LIST: i32 = 1000;
const MAX_DELETE_LIST: i32 = 1000;
const MAX_UPLOADS_LIST: i32 = 10000;
const MAX_PARTS_LIST: i32 = 10000;
const METACACHE_SHARE_PREFIX: bool = false;
fn max_keys_plus_one(max_keys: i32, add_one: bool) -> i32 {
let mut max_keys = max_keys;
if max_keys > MAX_OBJECT_LIST {
max_keys = MAX_OBJECT_LIST;
}
if add_one {
max_keys += 1;
}
max_keys
}
#[derive(Debug, Default, Clone)]
pub 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,
// The number of disks to ask.
pub ask_disks: String,
// InclDeleted will keep all entries where latest version is a delete marker.
pub incl_deleted: bool,
// Scan recursively.
// If false only main directory will be scanned.
// Should always be true if Separator is n SlashSeparator.
pub recursive: bool,
// Separator to use.
pub separator: String,
// Create indicates that the lister should not attempt to load an existing cache.
pub create: bool,
// Include pure directories.
pub include_directories: bool,
// Transient is set if the cache is transient due to an error or being a reserved bucket.
// This means the cache metadata will not be persisted on disk.
// A transient result will never be returned from the cache so knowing the list id is required.
pub transient: bool,
// Versioned is this a ListObjectVersions call.
pub versioned: bool,
}
impl ListPathOptions {
pub fn set_filter(&mut self) {
if METACACHE_SHARE_PREFIX {
return;
}
if self.prefix == self.base_dir {
return;
}
let s = SLASH_SEPARATOR.chars().next().unwrap_or_default();
self.filter_prefix = self.prefix.trim_start_matches(&self.base_dir).trim_matches(s).to_owned();
if self.filter_prefix.contains(s) {
self.filter_prefix = "".to_owned();
}
}
}
impl ECStore {
pub async fn inner_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 marker = {
if continuation_token.is_empty() {
start_after
} else {
continuation_token
}
};
self.list_objects_generic(bucket, prefix, marker, delimiter, max_keys).await?;
unimplemented!()
}
pub async fn list_objects_generic(
&self,
bucket: &str,
prefix: &str,
marker: &str,
delimiter: &str,
max_keys: i32,
) -> Result<ListObjectsInfo> {
let opts = ListPathOptions {
bucket: bucket.to_owned(),
prefix: prefix.to_owned(),
separator: delimiter.to_owned(),
limit: max_keys_plus_one(max_keys, !marker.is_empty()),
marker: marker.to_owned(),
incl_deleted: false,
ask_disks: "strict".to_owned(), //TODO: from config
..Default::default()
};
let merged = self.list_path(&opts).await?;
todo!()
}
async fn list_path(&self, o: &ListPathOptions) -> Result<ListObjectsInfo> {
check_list_objs_args(&o.bucket, &o.prefix, &o.marker)?;
// if opts.prefix.ends_with(SLASH_SEPARATOR) {
// return Err(Error::msg("eof"));
// }
let mut o = o.clone();
if o.marker < o.prefix {
o.marker = "".to_owned();
}
if !o.marker.is_empty() && !o.prefix.is_empty() {
if !o.marker.starts_with(&o.prefix) {
return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
}
}
if o.limit == 0 {
return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
}
if o.prefix.ends_with(SLASH_SEPARATOR) {
return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof)));
}
o.include_directories = o.separator == SLASH_SEPARATOR;
if (o.separator == SLASH_SEPARATOR || o.separator.is_empty()) && !o.recursive {
o.recursive = o.separator != SLASH_SEPARATOR;
o.separator = SLASH_SEPARATOR.to_owned();
} else {
o.recursive = true
}
// TODO: parseMarker
if o.base_dir.is_empty() {
o.base_dir = base_dir_from_prefix(&o.prefix);
}
o.transient = o.transient || is_reserved_or_invalid_bucket(&o.bucket, false);
o.set_filter();
if o.transient {
o.create = false;
}
todo!()
// let mut opts = opts.clone();
// if opts.base_dir.is_empty() {
// opts.base_dir = base_dir_from_prefix(&opts.prefix);
// }
// let objects = self.list_merged(&opts).await?;
// let info = ListObjectsInfo {
// objects,
// ..Default::default()
// };
// Ok(info)
}
// 读所有
async fn list_merged(&self, opts: &ListPathOptions) -> Result<Vec<ObjectInfo>> {
let walk_opts = WalkDirOptions {
bucket: opts.bucket.clone(),
base_dir: opts.base_dir.clone(),
..Default::default()
};
// let (mut wr, mut rd) = tokio::io::duplex(1024);
let mut futures = Vec::new();
for sets in self.pools.iter() {
for set in sets.disk_set.iter() {
futures.push(set.walk_dir(&walk_opts));
}
}
let results = join_all(futures).await;
// let mut errs = Vec::new();
let mut ress = Vec::new();
let mut uniq = HashSet::new();
for (disks_ress, _disks_errs) in results {
for disks_res in disks_ress.iter() {
if disks_res.is_none() {
// TODO handle errs
continue;
}
let entrys = disks_res.as_ref().unwrap();
for entry in entrys {
// warn!("lst_merged entry---- {}", &entry.name);
if !opts.prefix.is_empty() && !entry.name.starts_with(&opts.prefix) {
continue;
}
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() {
// if !opts.delimiter.is_empty() {
// // entry.name.trim_start_matches(pat)
// }
let fi = entry.to_fileinfo(&opts.bucket)?;
if let Some(f) = fi {
ress.push(f.to_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.clone(),
..Default::default()
});
}
}
}
}
}
// warn!("list_merged errs {:?}", errs);
Ok(ress)
}
}
ecstore/src/metacache/mod.rs
+1
View File
@@ -0,0 +1 @@
pub mod writer;
ecstore/src/metacache/writer.rs
+388
View File
@@ -0,0 +1,388 @@
use crate::disk::MetaCacheEntry;
use crate::error::Error;
use crate::error::Result;
use rmp::Marker;
use std::str::from_utf8;
use tokio::io::AsyncRead;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWrite;
use tokio::io::AsyncWriteExt;
// use std::sync::Arc;
// use tokio::sync::mpsc;
// use tokio::sync::mpsc::Sender;
// use tokio::task;
const METACACHE_STREAM_VERSION: u8 = 2;
#[derive(Debug)]
pub struct MetacacheWriter<W> {
wr: W,
created: bool,
// err: Option<Error>,
buf: Vec<u8>,
}
impl<W: AsyncWrite + Unpin> MetacacheWriter<W> {
pub fn new(wr: W) -> Self {
Self {
wr,
created: false,
// err: None,
buf: Vec::new(),
}
}
pub async fn flush(&mut self) -> Result<()> {
self.wr.write_all(&self.buf).await?;
self.buf.clear();
Ok(())
}
pub async fn init(&mut self) -> Result<()> {
if !self.created {
rmp::encode::write_u8(&mut self.buf, METACACHE_STREAM_VERSION).map_err(|e| Error::msg(format!("{:?}", e)))?;
self.flush().await?;
self.created = true;
}
Ok(())
}
pub async fn write(&mut self, objs: &[MetaCacheEntry]) -> Result<()> {
if objs.is_empty() {
return Ok(());
}
self.init().await?;
for obj in objs.iter() {
if obj.name.is_empty() {
return Err(Error::msg("metacacheWriter: no name"));
}
self.write_obj(obj).await?;
}
Ok(())
}
pub async fn write_obj(&mut self, obj: &MetaCacheEntry) -> Result<()> {
self.init().await?;
rmp::encode::write_bool(&mut self.buf, true).map_err(|e| Error::msg(format!("{:?}", e)))?;
rmp::encode::write_str(&mut self.buf, &obj.name).map_err(|e| Error::msg(format!("{:?}", e)))?;
rmp::encode::write_bin(&mut self.buf, &obj.metadata).map_err(|e| Error::msg(format!("{:?}", e)))?;
self.flush().await?;
Ok(())
}
// pub async fn stream(&mut self) -> Result<Sender<MetaCacheEntry>> {
// let (sender, mut receiver) = mpsc::channel::<MetaCacheEntry>(100);
// let wr = Arc::new(self);
// task::spawn(async move {
// while let Some(obj) = receiver.recv().await {
// // if obj.name.is_empty() || self.err.is_some() {
// // continue;
// // }
// let _ = wr.write_obj(&obj);
// // if let Err(err) = rmp::encode::write_bool(&mut self.wr, true) {
// // self.err = Some(Error::new(err));
// // continue;
// // }
// // if let Err(err) = rmp::encode::write_str(&mut self.wr, &obj.name) {
// // self.err = Some(Error::new(err));
// // continue;
// // }
// // if let Err(err) = rmp::encode::write_bin(&mut self.wr, &obj.metadata) {
// // self.err = Some(Error::new(err));
// // continue;
// // }
// }
// });
// Ok(sender)
// }
pub async fn close(&mut self) -> Result<()> {
rmp::encode::write_bool(&mut self.buf, false).map_err(|e| Error::msg(format!("{:?}", e)))?;
self.flush().await?;
Ok(())
}
}
pub struct MetacacheReader<R> {
rd: R,
init: bool,
err: Option<Error>,
buf: Vec<u8>,
offset: usize,
current: Option<MetaCacheEntry>,
}
impl<R: AsyncRead + Unpin> MetacacheReader<R> {
pub fn new(rd: R) -> Self {
Self {
rd,
init: false,
err: None,
buf: Vec::new(),
offset: 0,
current: None,
}
}
pub async fn read_more(&mut self, read_size: usize) -> Result<&[u8]> {
let ext_size = read_size + self.offset;
let extra = ext_size - self.offset;
if self.buf.capacity() >= ext_size {
// Extend the buffer if we have enough space.
self.buf.resize(ext_size, 0);
} else {
self.buf.extend(vec![0u8; extra]);
}
let pref = self.offset;
self.rd.read_exact(&mut self.buf[pref..ext_size]).await?;
self.offset += read_size;
let data = &self.buf[pref..ext_size];
Ok(data)
}
fn reset(&mut self) {
self.buf.clear();
self.offset = 0;
}
async fn check_init(&mut self) -> Result<()> {
if !self.init {
let ver = match rmp::decode::read_u8(&mut self.read_more(2).await?) {
Ok(res) => res,
Err(err) => {
self.err = Some(Error::msg(format!("{:?}", err)));
0
}
};
match ver {
1 | 2 => (),
_ => {
self.err = Some(Error::msg("invalid version"));
}
}
self.init = true;
}
Ok(())
}
async fn read_str_len(&mut self) -> Result<u32> {
let mark = match rmp::decode::read_marker(&mut self.read_more(1).await?) {
Ok(res) => res,
Err(err) => {
let serr = format!("{:?}", err);
self.err = Some(Error::msg(&serr));
return Err(Error::msg(&serr));
}
};
match mark {
Marker::FixStr(size) => Ok(u32::from(size)),
Marker::Str8 => Ok(u32::from(self.read_u8().await?)),
Marker::Str16 => Ok(u32::from(self.read_u16().await?)),
Marker::Str32 => Ok(self.read_u32().await?),
_marker => Err(Error::msg("str marker err")),
}
}
async fn read_bin_len(&mut self) -> Result<u32> {
let mark = match rmp::decode::read_marker(&mut self.read_more(1).await?) {
Ok(res) => res,
Err(err) => {
let serr = format!("{:?}", err);
self.err = Some(Error::msg(&serr));
return Err(Error::msg(&serr));
}
};
match mark {
Marker::Bin8 => Ok(u32::from(self.read_u8().await?)),
Marker::Bin16 => Ok(u32::from(self.read_u16().await?)),
Marker::Bin32 => Ok(self.read_u32().await?),
_ => Err(Error::msg("bin marker err")),
}
}
async fn read_u8(&mut self) -> Result<u8> {
let buf = self.read_more(1).await?;
Ok(u8::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
}
async fn read_u16(&mut self) -> Result<u16> {
let buf = self.read_more(2).await?;
Ok(u16::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
}
async fn read_u32(&mut self) -> Result<u32> {
let buf = self.read_more(4).await?;
Ok(u32::from_be_bytes(buf.try_into().expect("Slice with incorrect length")))
}
pub async fn skip(&mut self, size: usize) -> Result<()> {
self.check_init().await?;
if let Some(err) = &self.err {
return Err(err.clone());
}
let mut n = size;
if self.current.is_some() {
n -= 1;
self.current = None;
}
while n > 0 {
match rmp::decode::read_bool(&mut self.read_more(1).await?) {
Ok(res) => {
if !res {
return Ok(());
}
}
Err(err) => {
let serr = format!("{:?}", err);
self.err = Some(Error::msg(&serr));
return Err(Error::msg(&serr));
}
};
let l = self.read_str_len().await?;
let _ = self.read_more(l as usize).await?;
let l = self.read_bin_len().await?;
let _ = self.read_more(l as usize).await?;
n -= 1;
}
Ok(())
}
pub async fn peek(&mut self) -> Result<Option<MetaCacheEntry>> {
self.check_init().await?;
if let Some(err) = &self.err {
return Err(err.clone());
}
match rmp::decode::read_bool(&mut self.read_more(1).await?) {
Ok(res) => {
if !res {
return Ok(None);
}
}
Err(err) => {
let serr = format!("{:?}", err);
self.err = Some(Error::msg(&serr));
return Err(Error::msg(&serr));
}
};
let l = self.read_str_len().await?;
let buf = self.read_more(l as usize).await?;
let name_buf = buf.to_vec();
let name = match from_utf8(&name_buf) {
Ok(decoded) => decoded.to_owned(),
Err(err) => {
self.err = Some(Error::msg(err.to_string()));
return Err(Error::msg(err.to_string()));
}
};
let l = self.read_bin_len().await?;
let buf = self.read_more(l as usize).await?;
let metadata = buf.to_vec();
self.reset();
let entry = Some(MetaCacheEntry {
name,
metadata,
cached: None,
reusable: false,
});
self.current = entry.clone();
Ok(entry)
}
pub async fn read_all(&mut self) -> Result<Vec<MetaCacheEntry>> {
let mut ret = Vec::new();
loop {
if let Some(entry) = self.peek().await? {
ret.push(entry);
continue;
}
break;
}
Ok(ret)
}
}
#[tokio::test]
async fn test_writer() {
use crate::io::VecAsyncReader;
use crate::io::VecAsyncWriter;
let mut f = VecAsyncWriter::new(Vec::new());
let mut w = MetacacheWriter::new(&mut f);
let mut objs = Vec::new();
for i in 0..10 {
let info = MetaCacheEntry {
name: format!("item{}", i),
metadata: vec![0u8, 10],
cached: None,
reusable: false,
};
println!("old {:?}", &info);
objs.push(info);
}
w.write(&objs).await.unwrap();
w.close().await.unwrap();
let data = f.get_buffer().to_vec();
let nf = VecAsyncReader::new(data);
let mut r = MetacacheReader::new(nf);
let nobjs = r.read_all().await.unwrap();
for info in nobjs.iter() {
println!("new {:?}", &info);
}
assert_eq!(objs, nobjs)
}
ecstore/src/notification_sys.rs
+4
View File
@@ -26,6 +26,7 @@ pub fn get_global_notification_sys() -> Option<&'static NotificationSys> {
pub struct NotificationSys {
pub peer_clients: Vec<Option<PeerRestClient>>,
#[allow(dead_code)]
pub all_peer_clients: Vec<Option<PeerRestClient>>,
}
@@ -45,6 +46,9 @@ pub struct NotificationPeerErr {
}
impl NotificationSys {
pub fn rest_client_from_hash(&self, _s: &str) -> Option<PeerRestClient> {
None
}
pub async fn delete_policy(&self) -> Vec<NotificationPeerErr> {
unimplemented!()
}
ecstore/src/peer.rs
+1 -1
View File
@@ -9,7 +9,7 @@ use regex::Regex;
use std::{collections::HashMap, fmt::Debug, sync::Arc};
use tokio::sync::RwLock;
use tonic::Request;
use tracing::{info, warn};
use tracing::info;
use crate::disk::error::is_all_buckets_not_found;
use crate::disk::{DiskAPI, DiskStore};
ecstore/src/set_disk.rs
+154 -63
View File
@@ -20,8 +20,8 @@ use crate::{
format::FormatV3,
new_disk, BufferReader, BufferWriter, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskOption,
DiskStore, FileInfoVersions, FileReader, FileWriter, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams,
ReadMultipleReq, ReadMultipleResp, ReadOptions, UpdateMetadataOpts, WalkDirOptions, RUSTFS_META_BUCKET,
RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET,
ReadMultipleReq, ReadMultipleResp, ReadOptions, UpdateMetadataOpts, RUSTFS_META_BUCKET, RUSTFS_META_MULTIPART_BUCKET,
RUSTFS_META_TMP_BUCKET,
},
erasure::Erasure,
error::{Error, Result},
@@ -160,6 +160,89 @@ impl SetDisks {
disks
}
pub async fn get_online_disks_with_healing_and_info(&self, incl_healing: bool) -> (Vec<DiskStore>, Vec<DiskInfo>, usize) {
let mut disks = self.get_disks_internal().await;
let mut infos = Vec::with_capacity(disks.len());
let mut futures = Vec::with_capacity(disks.len());
let mut numbers: Vec<usize> = (0..disks.len()).collect();
{
let mut rng = thread_rng();
disks.shuffle(&mut rng);
numbers.shuffle(&mut rand::thread_rng());
}
for &i in numbers.iter() {
let disk = disks[i].clone();
futures.push(async move {
if let Some(disk) = disk {
disk.disk_info(&DiskInfoOptions::default()).await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
});
}
let results = join_all(futures).await;
for result in results {
match result {
Ok(res) => {
infos.push(res);
}
Err(err) => {
infos.push(DiskInfo {
error: err.to_string(),
..Default::default()
});
}
}
}
let mut healing: usize = 0;
let mut scanning_disks = Vec::new();
let mut healing_disks = Vec::new();
let mut scanning_infos = Vec::new();
let mut healing_infos = Vec::new();
let mut new_disks = Vec::new();
let mut new_infos = Vec::new();
for &i in numbers.iter() {
let (info, disk) = (infos[i].clone(), disks[i].clone());
if !info.error.is_empty() || disk.is_none() {
continue;
}
if info.healing {
healing += 1;
if incl_healing {
healing_disks.push(disk.unwrap());
healing_infos.push(info);
}
continue;
}
if !info.healing {
new_disks.push(disk.unwrap());
new_infos.push(info);
} else {
scanning_disks.push(disk.unwrap());
scanning_infos.push(info);
}
}
new_disks.extend(scanning_disks);
new_infos.extend(scanning_infos);
new_disks.extend(healing_disks);
new_infos.extend(healing_infos);
(new_disks, new_infos, healing)
}
async fn _get_local_disks(&self) -> Vec<Option<DiskStore>> {
let mut disks = self.get_disks_internal().await;
@@ -741,6 +824,7 @@ impl SetDisks {
};
if let Some(err) = reduce_read_quorum_errs(errs, object_op_ignored_errs().as_ref(), expected_rquorum) {
// warn!("object_quorum_from_meta err {:?}", &err);
return Err(err);
}
@@ -1117,9 +1201,9 @@ impl SetDisks {
let finfo = match meta.into_fileinfo(bucket, object, "", true, true) {
Ok(res) => res,
Err(err) => {
for i in 0..errs.len() {
if errs[i].is_none() {
errs[i] = Some(err.clone())
for item in errs.iter_mut() {
if item.is_none() {
*item = Some(err.clone())
}
}
@@ -1128,9 +1212,9 @@ impl SetDisks {
};
if !finfo.is_valid() {
for i in 0..errs.len() {
if errs[i].is_none() {
errs[i] = Some(Error::new(DiskError::FileCorrupt));
for item in errs.iter_mut() {
if item.is_none() {
*item = Some(Error::new(DiskError::FileCorrupt));
}
}
@@ -1333,42 +1417,42 @@ impl SetDisks {
Ok((disk, fm))
}
pub async fn walk_dir(&self, opts: &WalkDirOptions) -> (Vec<Option<Vec<MetaCacheEntry>>>, Vec<Option<Error>>) {
let disks = self.disks.read().await;
// pub async fn walk_dir(&self, opts: &WalkDirOptions) -> (Vec<Option<Vec<MetaCacheEntry>>>, Vec<Option<Error>>) {
// let disks = self.disks.read().await;
let disks = disks.clone();
let mut futures = Vec::new();
let mut errs = Vec::new();
let mut ress = Vec::new();
// let disks = disks.clone();
// let mut futures = Vec::new();
// let mut errs = Vec::new();
// let mut ress = Vec::new();
for disk in disks.iter() {
let opts = opts.clone();
futures.push(async move {
if let Some(disk) = disk {
disk.walk_dir(opts).await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
});
}
// for disk in disks.iter() {
// let opts = opts.clone();
// futures.push(async move {
// if let Some(disk) = disk {
// disk.walk_dir(opts, &mut Writer::NotUse).await
// } else {
// Err(Error::new(DiskError::DiskNotFound))
// }
// });
// }
let results = join_all(futures).await;
// let results = join_all(futures).await;
for res in results {
match res {
Ok(entrys) => {
ress.push(Some(entrys));
errs.push(None);
}
Err(e) => {
ress.push(None);
errs.push(Some(e));
}
}
}
// for res in results {
// match res {
// Ok(entrys) => {
// ress.push(Some(entrys));
// errs.push(None);
// }
// Err(e) => {
// ress.push(None);
// errs.push(Some(e));
// }
// }
// }
(ress, errs)
}
// (ress, errs)
// }
async fn remove_object_part(
&self,
@@ -1624,7 +1708,8 @@ impl SetDisks {
let _min_disks = self.set_drive_count - self.default_parity_count;
let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)?;
let (read_quorum, _) = Self::object_quorum_from_meta(&parts_metadata, &errs, self.default_parity_count)
.map_err(|err| to_object_err(err, vec![bucket, object]))?;
if let Some(err) = reduce_read_quorum_errs(&errs, object_op_ignored_errs().as_ref(), read_quorum as usize) {
error!(
@@ -1633,7 +1718,7 @@ impl SetDisks {
read_quorum,
&errs
);
return Err(err);
return Err(to_object_err(err, vec![bucket, object]));
}
let (op_online_disks, mot_time, etag) = Self::list_online_disks(&disks, &parts_metadata, &errs, read_quorum as usize);
@@ -1879,9 +1964,15 @@ impl SetDisks {
fallback_disks: fallback_disks.to_vec(),
bucket: bucket.to_string(),
path: path.to_string(),
filter_prefix: filter_prefix.to_string(),
filter_prefix: {
if filter_prefix.is_empty() {
None
} else {
Some(filter_prefix.to_string())
}
},
recursice: true,
forward_to: "".to_string(),
forward_to: None,
min_disks: 1,
report_not_found: false,
per_disk_limit: 0,
@@ -3039,10 +3130,10 @@ impl SetDisks {
continue;
}
let mut forward_to = "".to_string();
let mut forward_to = None;
let b = tracker.read().await.get_bucket().await;
if b == *bucket {
forward_to = tracker.read().await.get_object().await;
forward_to = Some(tracker.read().await.get_object().await);
}
if !b.is_empty() {
@@ -3811,24 +3902,24 @@ impl StorageAPI for SetDisks {
}
async fn list_objects_v2(
&self,
self: Arc<Self>,
_bucket: &str,
_prefix: &str,
_continuation_token: &str,
_delimiter: &str,
_continuation_token: Option<String>,
_delimiter: Option<String>,
_max_keys: i32,
_fetch_owner: bool,
_start_after: &str,
_start_after: Option<String>,
) -> Result<ListObjectsV2Info> {
unimplemented!()
}
async fn list_object_versions(
&self,
self: Arc<Self>,
_bucket: &str,
_prefix: &str,
_marker: &str,
_version_marker: &str,
_delimiter: &str,
_marker: Option<String>,
_version_marker: Option<String>,
_delimiter: Option<String>,
_max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
unimplemented!()
@@ -4092,9 +4183,9 @@ impl StorageAPI for SetDisks {
&self,
bucket: &str,
object: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
key_marker: Option<String>,
upload_id_marker: Option<String>,
delimiter: Option<String>,
max_uploads: usize,
) -> Result<ListMultipartsInfo> {
let disks = {
@@ -4188,14 +4279,14 @@ impl StorageAPI for SetDisks {
uploads.sort_by(|a, b| a.initiated.cmp(&b.initiated));
let mut upload_idx = 0;
if !upload_id_marker.is_empty() {
if let Some(upload_id_marker) = &upload_id_marker {
while upload_idx < uploads.len() {
if uploads[upload_idx].upload_id != upload_id_marker {
if &uploads[upload_idx].upload_id != upload_id_marker {
upload_idx += 1;
continue;
}
if uploads[upload_idx].upload_id == upload_id_marker {
if &uploads[upload_idx].upload_id == upload_id_marker {
upload_idx += 1;
break;
}
@@ -4205,10 +4296,10 @@ impl StorageAPI for SetDisks {
}
let mut ret_uploads = Vec::new();
let mut next_upload_id_marker = String::new();
let mut next_upload_id_marker = None;
while upload_idx < uploads.len() {
ret_uploads.push(uploads[upload_idx].clone());
next_upload_id_marker = uploads[upload_idx].upload_id.clone();
next_upload_id_marker = Some(uploads[upload_idx].upload_id.clone());
upload_idx += 1;
if ret_uploads.len() > max_uploads {
@@ -4219,7 +4310,7 @@ impl StorageAPI for SetDisks {
let is_truncated = ret_uploads.len() < uploads.len();
if !is_truncated {
next_upload_id_marker = "".to_owned();
next_upload_id_marker = None;
}
Ok(ListMultipartsInfo {
@@ -5125,7 +5216,7 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ma
ret
}
async fn get_storage_info(disks: &Vec<Option<DiskStore>>, eps: &Vec<Endpoint>) -> madmin::StorageInfo {
async fn get_storage_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> madmin::StorageInfo {
let mut disks = get_disks_info(disks, eps).await;
disks.sort_by(|a, b| a.total_space.cmp(&b.total_space));
ecstore/src/sets.rs
+13 -13
View File
@@ -87,7 +87,7 @@ impl Sets {
let mut disk_set = Vec::with_capacity(set_count);
for i in 0..set_count {
for (i, locker) in lockers.iter().enumerate().take(set_count) {
let mut set_drive = Vec::with_capacity(set_drive_count);
let mut set_endpoints = Vec::with_capacity(set_drive_count);
for j in 0..set_drive_count {
@@ -145,7 +145,7 @@ impl Sets {
// warn!("sets new set_drive {:?}", &set_drive);
let set_disks = SetDisks {
lockers: lockers[i].clone(),
lockers: locker.clone(),
locker_owner: GLOBAL_Local_Node_Name.read().await.to_string(),
ns_mutex: Arc::new(RwLock::new(NsLockMap::new(is_dist_erasure().await))),
disks: RwLock::new(set_drive),
@@ -451,24 +451,24 @@ impl StorageAPI for Sets {
self.get_disks_by_key(object).delete_object(bucket, object, opts).await
}
async fn list_objects_v2(
&self,
self: Arc<Self>,
_bucket: &str,
_prefix: &str,
_continuation_token: &str,
_delimiter: &str,
_continuation_token: Option<String>,
_delimiter: Option<String>,
_max_keys: i32,
_fetch_owner: bool,
_start_after: &str,
_start_after: Option<String>,
) -> Result<ListObjectsV2Info> {
unimplemented!()
}
async fn list_object_versions(
&self,
self: Arc<Self>,
_bucket: &str,
_prefix: &str,
_marker: &str,
_version_marker: &str,
_delimiter: &str,
_marker: Option<String>,
_version_marker: Option<String>,
_delimiter: Option<String>,
_max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
unimplemented!()
@@ -527,9 +527,9 @@ impl StorageAPI for Sets {
&self,
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
key_marker: Option<String>,
upload_id_marker: Option<String>,
delimiter: Option<String>,
max_uploads: usize,
) -> Result<ListMultipartsInfo> {
self.get_disks_by_key(prefix)
ecstore/src/store.rs
+153 -180
View File
@@ -24,19 +24,19 @@ use crate::store_err::{
};
use crate::store_init::ec_drives_no_config;
use crate::utils::crypto::base64_decode;
use crate::utils::path::{base_dir_from_prefix, decode_dir_object, encode_dir_object, SLASH_SEPARATOR};
use crate::utils::path::{decode_dir_object, encode_dir_object, SLASH_SEPARATOR};
use crate::utils::xml;
use crate::{
bucket::metadata::BucketMetadata,
disk::{error::DiskError, new_disk, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
disk::{error::DiskError, new_disk, DiskOption, DiskStore, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
endpoints::EndpointServerPools,
error::{Error, Result},
peer::S3PeerSys,
sets::Sets,
store_api::{
BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec,
ListObjectsInfo, ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete,
PartInfo, PutObjReader, StorageAPI,
ListObjectsV2Info, MakeBucketOptions, MultipartUploadResult, ObjectInfo, ObjectOptions, ObjectToDelete, PartInfo,
PutObjReader, StorageAPI,
},
store_init, utils,
};
@@ -52,11 +52,7 @@ use std::cmp::Ordering;
use std::process::exit;
use std::slice::Iter;
use std::time::SystemTime;
use std::{
collections::{HashMap, HashSet},
sync::Arc,
time::Duration,
};
use std::{collections::HashMap, sync::Arc, time::Duration};
use time::OffsetDateTime;
use tokio::select;
use tokio::sync::mpsc::Sender;
@@ -265,102 +261,104 @@ impl ECStore {
self.pools.len() == 1
}
pub async fn list_path(&self, opts: &ListPathOptions, delimiter: &str) -> Result<ListObjectsInfo> {
// if opts.prefix.ends_with(SLASH_SEPARATOR) {
// return Err(Error::msg("eof"));
// }
// define in store_list_objects.rs
// pub async fn list_path(&self, opts: &ListPathOptions, delimiter: &str) -> Result<ListObjectsInfo> {
// // if opts.prefix.ends_with(SLASH_SEPARATOR) {
// // return Err(Error::msg("eof"));
// // }
let mut opts = opts.clone();
// let mut opts = opts.clone();
if opts.base_dir.is_empty() {
opts.base_dir = base_dir_from_prefix(&opts.prefix);
}
// if opts.base_dir.is_empty() {
// opts.base_dir = base_dir_from_prefix(&opts.prefix);
// }
let objects = self.list_merged(&opts, delimiter).await?;
// let objects = self.list_merged(&opts, delimiter).await?;
let info = ListObjectsInfo {
objects,
..Default::default()
};
Ok(info)
}
// let info = ListObjectsInfo {
// objects,
// ..Default::default()
// };
// Ok(info)
// }
// 读所有
async fn list_merged(&self, opts: &ListPathOptions, delimiter: &str) -> Result<Vec<ObjectInfo>> {
let walk_opts = WalkDirOptions {
bucket: opts.bucket.clone(),
base_dir: opts.base_dir.clone(),
..Default::default()
};
// define in store_list_objects.rs
// async fn list_merged(&self, opts: &ListPathOptions, delimiter: &str) -> Result<Vec<ObjectInfo>> {
// let walk_opts = WalkDirOptions {
// bucket: opts.bucket.clone(),
// base_dir: opts.base_dir.clone(),
// ..Default::default()
// };
// let (mut wr, mut rd) = tokio::io::duplex(1024);
// // let (mut wr, mut rd) = tokio::io::duplex(1024);
let mut futures = Vec::new();
// let mut futures = Vec::new();
for sets in self.pools.iter() {
for set in sets.disk_set.iter() {
futures.push(set.walk_dir(&walk_opts));
}
}
// for sets in self.pools.iter() {
// for set in sets.disk_set.iter() {
// futures.push(set.walk_dir(&walk_opts));
// }
// }
let results = join_all(futures).await;
// let results = join_all(futures).await;
// let mut errs = Vec::new();
let mut ress = Vec::new();
let mut uniq = HashSet::new();
// // let mut errs = Vec::new();
// let mut ress = Vec::new();
// let mut uniq = HashSet::new();
for (disks_ress, _disks_errs) in results {
for disks_res in disks_ress.iter() {
if disks_res.is_none() {
// TODO handle errs
continue;
}
let entrys = disks_res.as_ref().unwrap();
// for (disks_ress, _disks_errs) in results {
// for disks_res in disks_ress.iter() {
// if disks_res.is_none() {
// // TODO handle errs
// continue;
// }
// let entrys = disks_res.as_ref().unwrap();
for entry in entrys {
// warn!("lst_merged entry---- {}", &entry.name);
// for entry in entrys {
// // warn!("lst_merged entry---- {}", &entry.name);
if !opts.prefix.is_empty() && !entry.name.starts_with(&opts.prefix) {
continue;
}
// if !opts.prefix.is_empty() && !entry.name.starts_with(&opts.prefix) {
// continue;
// }
if !uniq.contains(&entry.name) {
uniq.insert(entry.name.clone());
// TODO: 过滤
// 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 opts.limit > 0 && ress.len() as i32 >= opts.limit {
// return Ok(ress);
// }
if entry.is_object() {
if !delimiter.is_empty() {
// entry.name.trim_start_matches(pat)
}
// if entry.is_object() {
// if !delimiter.is_empty() {
// // entry.name.trim_start_matches(pat)
// }
let fi = entry.to_fileinfo(&opts.bucket)?;
if let Some(f) = fi {
ress.push(f.to_object_info(&opts.bucket, &entry.name, false));
}
continue;
}
// let fi = entry.to_fileinfo(&opts.bucket)?;
// if let Some(f) = fi {
// ress.push(f.to_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.clone(),
..Default::default()
});
}
}
}
}
}
// if entry.is_dir() {
// ress.push(ObjectInfo {
// is_dir: true,
// bucket: opts.bucket.clone(),
// name: entry.name.clone(),
// ..Default::default()
// });
// }
// }
// }
// }
// }
// warn!("list_merged errs {:?}", errs);
// // warn!("list_merged errs {:?}", errs);
Ok(ress)
}
// Ok(ress)
// }
async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> {
let mut futures = Vec::new();
@@ -1020,32 +1018,32 @@ pub struct PoolObjInfo {
pub err: Option<Error>,
}
#[derive(Debug, Default, Clone)]
pub struct ListPathOptions {
pub id: String,
// #[derive(Debug, Default, Clone)]
// pub struct ListPathOptions {
// pub id: String,
// Bucket of the listing.
pub bucket: 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,
// // 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,
// // 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,
// // 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,
// // 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,
}
// // Limit the number of results.
// pub limit: i32,
// }
#[async_trait::async_trait]
impl ObjectIO for ECStore {
@@ -1514,70 +1512,34 @@ impl StorageAPI for ECStore {
Err(Error::new(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned())))
}
// TODO: review
// @continuation_token marker
// @start_after as marker when continuation_token empty
// @delimiter default="/", empty when recursive
// @max_keys limit
async fn list_objects_v2(
&self,
self: Arc<Self>,
bucket: &str,
prefix: &str,
continuation_token: &str,
delimiter: &str,
continuation_token: Option<String>,
delimiter: Option<String>,
max_keys: i32,
_fetch_owner: bool,
_start_after: &str,
fetch_owner: bool,
start_after: Option<String>,
) -> Result<ListObjectsV2Info> {
let opts = ListPathOptions {
bucket: bucket.to_string(),
limit: max_keys,
prefix: prefix.to_owned(),
..Default::default()
};
let info = self.list_path(&opts, delimiter).await?;
// 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)
self.inner_list_objects_v2(bucket, prefix, continuation_token, delimiter, max_keys, fetch_owner, start_after)
.await
}
async fn list_object_versions(
&self,
_bucket: &str,
_prefix: &str,
marker: &str,
version_marker: &str,
_delimiter: &str,
_max_keys: i32,
self: Arc<Self>,
bucket: &str,
prefix: &str,
marker: Option<String>,
version_marker: Option<String>,
delimiter: Option<String>,
max_keys: i32,
) -> Result<ListObjectVersionsInfo> {
if marker.is_empty() && !version_marker.is_empty() {
return Err(Error::new(StorageError::NotImplemented));
}
// let opts = ListPathOptions {
// bucket: bucket.to_owned(),
// marker: marker.to_owned(),
// prefix: prefix.to_owned(),
// limit: max_keys,
// ..Default::default()
// };
// let list = self
// .list_path(&opts, delimiter)
// .await
// .map_err(|e| to_object_err(e, vec![bucket]))?;
// for info in list.objects.iter() {
// //
// }
// FIXME:
unimplemented!()
self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys)
.await
}
async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<ObjectInfo> {
check_object_args(bucket, object)?;
@@ -1711,12 +1673,12 @@ impl StorageAPI for ECStore {
&self,
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
key_marker: Option<String>,
upload_id_marker: Option<String>,
delimiter: Option<String>,
max_uploads: usize,
) -> Result<ListMultipartsInfo> {
check_list_multipart_args(bucket, prefix, key_marker, upload_id_marker, delimiter)?;
check_list_multipart_args(bucket, prefix, &key_marker, &upload_id_marker, &delimiter)?;
if prefix.is_empty() {
// TODO: return from cache
@@ -1732,14 +1694,21 @@ impl StorageAPI for ECStore {
for pool in self.pools.iter() {
let res = pool
.list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads)
.list_multipart_uploads(
bucket,
prefix,
key_marker.clone(),
upload_id_marker.clone(),
delimiter.clone(),
max_uploads,
)
.await?;
uploads.extend(res.uploads);
}
Ok(ListMultipartsInfo {
key_marker: key_marker.to_owned(),
upload_id_marker: upload_id_marker.to_owned(),
key_marker,
upload_id_marker,
max_uploads,
uploads,
prefix: prefix.to_owned(),
@@ -1757,7 +1726,7 @@ impl StorageAPI for ECStore {
for (idx, pool) in self.pools.iter().enumerate() {
// // TODO: IsSuspended
let res = pool
.list_multipart_uploads(bucket, object, "", "", "", MAX_UPLOADS_LIST)
.list_multipart_uploads(bucket, object, None, None, None, MAX_UPLOADS_LIST)
.await?;
if !res.uploads.is_empty() {
@@ -2174,9 +2143,11 @@ async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, po
*GLOBAL_Local_Node_Name.write().await = peer_set[0].clone();
}
pub fn is_valid_object_prefix(object: &str) -> bool {
pub fn is_valid_object_prefix(_object: &str) -> bool {
// Implement object prefix validation
!object.is_empty() // Placeholder
// !object.is_empty() // Placeholder
// FIXME: TODO:
true
}
fn is_valid_object_name(object: &str) -> bool {
@@ -2243,7 +2214,7 @@ fn check_bucket_and_object_names(bucket: &str, object: &str) -> Result<()> {
Ok(())
}
fn check_list_objs_args(bucket: &str, prefix: &str, _marker: &str) -> Result<()> {
pub fn check_list_objs_args(bucket: &str, prefix: &str, _marker: &Option<String>) -> Result<()> {
if !is_meta_bucketname(bucket) && check_valid_bucket_name_strict(bucket).is_err() {
return Err(Error::new(StorageError::BucketNameInvalid(bucket.to_string())));
}
@@ -2258,18 +2229,20 @@ fn check_list_objs_args(bucket: &str, prefix: &str, _marker: &str) -> Result<()>
fn check_list_multipart_args(
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
_delimiter: &str,
key_marker: &Option<String>,
upload_id_marker: &Option<String>,
_delimiter: &Option<String>,
) -> Result<()> {
check_list_objs_args(bucket, prefix, key_marker)?;
if !upload_id_marker.is_empty() {
if key_marker.ends_with('/') {
return Err(Error::new(StorageError::InvalidUploadIDKeyCombination(
upload_id_marker.to_string(),
key_marker.to_string(),
)));
if let Some(upload_id_marker) = upload_id_marker {
if let Some(key_marker) = key_marker {
if key_marker.ends_with('/') {
return Err(Error::new(StorageError::InvalidUploadIDKeyCombination(
upload_id_marker.to_string(),
key_marker.to_string(),
)));
}
}
if let Err(_e) = base64_decode(upload_id_marker.as_bytes()) {
ecstore/src/store_api.rs
+21 -21
View File
@@ -690,7 +690,7 @@ pub struct ListObjectsInfo {
// 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,
pub next_marker: Option<String>,
// List of objects info for this request.
pub objects: Vec<ObjectInfo>,
@@ -713,8 +713,8 @@ pub struct ListObjectsV2Info {
//
// NOTE: This element is returned only if you have delimiter request parameter
// specified.
pub continuation_token: String,
pub next_continuation_token: String,
pub continuation_token: Option<String>,
pub next_continuation_token: Option<String>,
// List of objects info for this request.
pub objects: Vec<ObjectInfo>,
@@ -746,20 +746,20 @@ pub struct MultipartInfo {
pub struct ListMultipartsInfo {
// Together with upload-id-marker, this parameter specifies the multipart upload
// after which listing should begin.
pub key_marker: String,
pub key_marker: Option<String>,
// Together with key-marker, specifies the multipart upload after which listing
// should begin. If key-marker is not specified, the upload-id-marker parameter
// is ignored.
pub upload_id_marker: String,
pub upload_id_marker: Option<String>,
// When a list is truncated, this element specifies the value that should be
// used for the key-marker request parameter in a subsequent request.
pub next_key_marker: String,
pub next_key_marker: Option<String>,
// When a list is truncated, this element specifies the value that should be
// used for the upload-id-marker request parameter in a subsequent request.
pub next_upload_id_marker: String,
pub next_upload_id_marker: Option<String>,
// Maximum number of multipart uploads that could have been included in the
// response.
@@ -780,7 +780,7 @@ pub struct ListMultipartsInfo {
// A character used to truncate the object prefixes.
// NOTE: only supported delimiter is '/'.
pub delimiter: String,
pub delimiter: Option<String>,
// CommonPrefixes contains all (if there are any) keys between Prefix and the
// next occurrence of the string specified by delimiter.
@@ -807,8 +807,8 @@ pub struct DeletedObject {
pub struct ListObjectVersionsInfo {
pub is_truncated: bool,
pub next_marker: String,
pub next_version_idmarker: String,
pub next_marker: Option<String>,
pub next_version_idmarker: Option<String>,
pub objects: Vec<ObjectInfo>,
pub prefixes: Vec<String>,
}
@@ -845,23 +845,23 @@ pub trait StorageAPI: ObjectIO {
async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()>;
// ListObjects TODO: FIXME:
async fn list_objects_v2(
&self,
self: Arc<Self>,
bucket: &str,
prefix: &str,
continuation_token: &str,
delimiter: &str,
continuation_token: Option<String>,
delimiter: Option<String>,
max_keys: i32,
fetch_owner: bool,
start_after: &str,
start_after: Option<String>,
) -> Result<ListObjectsV2Info>;
// ListObjectVersions TODO: FIXME:
async fn list_object_versions(
&self,
self: Arc<Self>,
bucket: &str,
prefix: &str,
marker: &str,
version_marker: &str,
delimiter: &str,
marker: Option<String>,
version_marker: Option<String>,
delimiter: Option<String>,
max_keys: i32,
) -> Result<ListObjectVersionsInfo>;
// Walk TODO:
@@ -886,9 +886,9 @@ pub trait StorageAPI: ObjectIO {
&self,
bucket: &str,
prefix: &str,
key_marker: &str,
upload_id_marker: &str,
delimiter: &str,
key_marker: Option<String>,
upload_id_marker: Option<String>,
delimiter: Option<String>,
max_uploads: usize,
) -> Result<ListMultipartsInfo>;
async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<MultipartUploadResult>;
ecstore/src/store_err.rs
+13
View File
@@ -52,6 +52,9 @@ pub enum StorageError {
#[error("Object not found: {0}/{1}")]
ObjectNotFound(String, String),
#[error("volume not found: {0}")]
VolumeNotFound(String),
#[error("Version not found: {0}/{1}-{2}")]
VersionNotFound(String, String, String),
@@ -107,6 +110,7 @@ impl StorageError {
StorageError::InsufficientWriteQuorum => 0x17,
StorageError::DecommissionNotStarted => 0x18,
StorageError::InvalidPart(_, _, _) => 0x19,
StorageError::VolumeNotFound(_) => 0x20,
}
}
@@ -141,6 +145,7 @@ impl StorageError {
0x17 => Some(StorageError::InsufficientWriteQuorum),
0x18 => Some(StorageError::DecommissionNotStarted),
0x19 => Some(StorageError::InvalidPart(Default::default(), Default::default(), Default::default())),
0x20 => Some(StorageError::VolumeNotFound(Default::default())),
_ => None,
}
}
@@ -263,6 +268,14 @@ pub fn is_err_bucket_exists(err: &Error) -> bool {
}
}
pub fn is_err_bucket_not_found(err: &Error) -> bool {
if let Some(e) = err.downcast_ref::<StorageError>() {
matches!(e, StorageError::VolumeNotFound(_)) || matches!(e, StorageError::BucketNotFound(_))
} else {
false
}
}
pub fn is_err_object_not_found(err: &Error) -> bool {
if is_err_file_not_found(err) {
return true;
ecstore/src/store_list_objects.rs
+1349
View File
File diff suppressed because it is too large Load Diff
ecstore/src/utils/path.rs
+28 -1
View File
@@ -1,6 +1,7 @@
use std::path::Path;
use std::path::PathBuf;
const GLOBAL_DIR_SUFFIX: &str = "__XLDIR__";
pub const GLOBAL_DIR_SUFFIX: &str = "__XLDIR__";
pub const SLASH_SEPARATOR: &str = "/";
@@ -69,6 +70,32 @@ pub fn path_join(elem: &[PathBuf]) -> PathBuf {
joined_path
}
pub fn path_join_buf(elements: &[&str]) -> String {
let trailing_slash = !elements.is_empty() && elements.last().unwrap().ends_with('/');
let mut dst = String::new();
let mut added = 0;
for e in elements {
if added > 0 || !e.is_empty() {
if added > 0 {
dst.push('/');
}
dst.push_str(e);
added += e.len();
}
}
let result = dst.to_string();
let cpath = Path::new(&result).components().collect::<PathBuf>();
let clean_path = cpath.to_string_lossy();
if trailing_slash {
return format!("{}/", clean_path);
}
clean_path.to_string()
}
pub fn path_to_bucket_object_with_base_path(bash_path: &str, path: &str) -> (String, String) {
let path = path.trim_start_matches(bash_path).trim_start_matches(SLASH_SEPARATOR);
if let Some(m) = path.find(SLASH_SEPARATOR) {
iam/src/manager.rs
+2 -1
View File
@@ -7,6 +7,7 @@ use std::{
time::Duration,
};
use ecstore::store_err::is_err_object_not_found;
use log::debug;
use time::OffsetDateTime;
use tokio::{
@@ -113,7 +114,7 @@ where
async fn save_iam_formatter(self: Arc<Self>) -> crate::Result<()> {
match self.api.load_iam_config::<Format>(Format::PATH).await {
Ok((format, _)) if format.version >= 1 => return Ok(()),
Err(Error::EcstoreError(e)) if !ecstore::disk::error::is_err_file_not_found(&e) => {
Err(Error::EcstoreError(e)) if !is_err_object_not_found(&e) => {
return Err(Error::EcstoreError(e));
}
_ => {}
iam/src/store/object.rs
+24 -20
View File
@@ -2,14 +2,14 @@ use std::{collections::HashMap, path::Path, sync::Arc};
use ecstore::{
config::error::is_not_found,
store::{ECStore, ListPathOptions},
store::ECStore,
store_api::{HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader},
utils::path::dir,
StorageAPI,
};
use futures::future::try_join_all;
use log::{debug, warn};
use serde::{de::DeserializeOwned, Serialize};
use tracing::error;
use super::Store;
use crate::{
@@ -40,32 +40,36 @@ impl ObjectStore {
for item in items {
let prefix = format!("{}{}", prefix, item);
futures.push(async move {
// let items = self
// .object_api
// .clone()
// .list_path(&ListPathOptions {
// bucket: Self::BUCKET_NAME.into(),
// prefix: prefix.clone(),
// ..Default::default()
// })
// .await;
let items = self
.object_api
.list_path(
&ListPathOptions {
bucket: Self::BUCKET_NAME.into(),
prefix: prefix.clone(),
..Default::default()
},
"",
)
.clone()
.list_objects_v2(Self::BUCKET_NAME, &prefix.clone(), None, None, 0, false, None)
.await;
match items {
Ok(items) => Result::<_, crate::Error>::Ok(items.objects),
Err(e) if is_not_found(&e) => Result::<_, crate::Error>::Ok(vec![]),
Err(e) => Err(Error::StringError(format!("list {prefix} failed, err: {e:?}"))),
Ok(items) => Result::<_, crate::Error>::Ok(items.prefixes),
Err(e) => {
if is_not_found(&e) {
Result::<_, crate::Error>::Ok(vec![])
} else {
Err(Error::StringError(format!("list {prefix} failed, err: {e:?}")))
}
}
}
});
}
Ok(try_join_all(futures)
.await?
.into_iter()
.flat_map(|x| x.into_iter())
.map(|x| x.name)
.collect())
// TODO: FIXME:
Ok(try_join_all(futures).await?.into_iter().flat_map(|x| x.into_iter()).collect())
}
async fn load_policy(&self, name: &str) -> crate::Result<PolicyDoc> {
rustfs/src/grpc.rs
+68 -44
View File
@@ -10,7 +10,7 @@ use ecstore::{
bucket::{metadata::load_bucket_metadata, metadata_sys},
disk::{
DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, Reader,
UpdateMetadataOpts, WalkDirOptions,
UpdateMetadataOpts,
},
erasure::Writer,
error::Error as EcsError,
@@ -23,13 +23,16 @@ use ecstore::{
peer::{LocalPeerS3Client, PeerS3Client},
store::{all_local_disk_path, find_local_disk},
store_api::{BucketOptions, DeleteBucketOptions, FileInfo, MakeBucketOptions, StorageAPI},
store_err::StorageError,
utils::err_to_proto_err,
};
use futures::{Stream, StreamExt};
use futures_util::future::join_all;
use lock::{lock_args::LockArgs, Locker, GLOBAL_LOCAL_SERVER};
use common::globals::GLOBAL_Local_Node_Name;
use ecstore::store_err::StorageError;
use ecstore::utils::err_to_proto_err;
use ecstore::disk::error::is_err_eof;
use ecstore::metacache::writer::MetacacheReader;
use madmin::health::{
get_cpus, get_mem_info, get_os_info, get_partitions, get_proc_info, get_sys_config, get_sys_errors, get_sys_services,
};
@@ -40,6 +43,7 @@ use protos::{
};
use rmp_serde::{Deserializer, Serializer};
use serde::{Deserialize, Serialize};
use tokio::spawn;
use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;
use tonic::{Request, Response, Status, Streaming};
@@ -819,54 +823,74 @@ impl Node for NodeService {
}
}
async fn walk_dir(&self, request: Request<WalkDirRequest>) -> Result<Response<WalkDirResponse>, Status> {
type WalkDirStream = ResponseStream<WalkDirResponse>;
async fn walk_dir(&self, request: Request<WalkDirRequest>) -> Result<Response<Self::WalkDirStream>, Status> {
info!("walk_dir");
let request = request.into_inner();
let (tx, rx) = mpsc::channel(128);
if let Some(disk) = self.find_disk(&request.disk).await {
let opts = match serde_json::from_str::<WalkDirOptions>(&request.walk_dir_options) {
let mut buf = Deserializer::new(Cursor::new(request.walk_dir_options));
let opts = match Deserialize::deserialize(&mut buf) {
Ok(options) => options,
Err(err) => {
return Ok(tonic::Response::new(WalkDirResponse {
success: false,
meta_cache_entry: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(
Default::default(),
Default::default(),
Default::default(),
)),
&format!("decode WalkDirOptions failed: {}", err),
)),
}));
Err(_) => {
return Err(Status::invalid_argument("invalid WalkDirOptions"));
}
};
match disk.walk_dir(opts).await {
Ok(entries) => {
let entries = entries
.into_iter()
.filter_map(|entry| serde_json::to_string(&entry).ok())
.collect();
Ok(tonic::Response::new(WalkDirResponse {
success: true,
meta_cache_entry: entries,
error: None,
}))
}
Err(err) => Ok(tonic::Response::new(WalkDirResponse {
success: false,
meta_cache_entry: Vec::new(),
error: Some(err_to_proto_err(&err, &format!("walk dir failed: {}", err))),
})),
}
spawn(async {
let (rd, mut wr) = tokio::io::duplex(64);
let job1 = spawn(async move {
if let Err(err) = disk.walk_dir(opts, &mut wr).await {
println!("walk_dir err {:?}", err);
}
});
let job2 = spawn(async move {
let mut reader = MetacacheReader::new(rd);
loop {
match reader.peek().await {
Ok(res) => {
if let Some(info) = res {
match serde_json::to_string(&info) {
Ok(meta_cache_entry) => tx
.send(Ok(WalkDirResponse {
success: true,
meta_cache_entry,
error_info: None,
}))
.await
.expect("working rx"),
Err(e) => tx
.send(Ok(WalkDirResponse {
success: false,
meta_cache_entry: "".to_string(),
error_info: Some(e.to_string()),
}))
.await
.expect("working rx"),
}
} else {
break;
}
}
Err(err) => {
if is_err_eof(&err) {
break;
}
println!("get err {:?}", err);
break;
}
}
}
});
join_all(vec![job1, job2]).await;
});
} else {
Ok(tonic::Response::new(WalkDirResponse {
success: false,
meta_cache_entry: Vec::new(),
error: Some(err_to_proto_err(
&EcsError::new(StorageError::InvalidArgument(Default::default(), Default::default(), Default::default())),
"can not find disk",
)),
}))
return Err(Status::invalid_argument(format!("invalid disk, all disk: {:?}", self.all_disk().await)));
}
let out_stream = ReceiverStream::new(rx);
Ok(tonic::Response::new(Box::pin(out_stream)))
}
async fn rename_data(&self, request: Request<RenameDataRequest>) -> Result<Response<RenameDataResponse>, Status> {
rustfs/src/storage/ecfs.rs
+87 -11
View File
@@ -467,8 +467,7 @@ impl S3 for FS {
#[tracing::instrument(level = "debug", skip(self, req))]
async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
// warn!("list_objects_v2 input {:?}", &req.input);
// warn!("list_objects_v2 req {:?}", &req.input);
let ListObjectsV2Input {
bucket,
continuation_token,
@@ -481,7 +480,11 @@ impl S3 for FS {
} = req.input;
let prefix = prefix.unwrap_or_default();
let delimiter = delimiter.unwrap_or_default();
let max_keys = max_keys.unwrap_or(1000);
let delimiter = delimiter.filter(|v| !v.is_empty());
let continuation_token = continuation_token.filter(|v| !v.is_empty());
let start_after = start_after.filter(|v| !v.is_empty());
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
@@ -491,16 +494,16 @@ impl S3 for FS {
.list_objects_v2(
&bucket,
&prefix,
&continuation_token.unwrap_or_default(),
&delimiter,
max_keys.unwrap_or_default(),
continuation_token,
delimiter.clone(),
max_keys,
fetch_owner.unwrap_or_default(),
&start_after.unwrap_or_default(),
start_after,
)
.await
.map_err(to_s3_error)?;
// warn!("object_infos {:?}", object_infos);
// warn!("object_infos objects {:?}", object_infos.objects);
let objects: Vec<Object> = object_infos
.objects
@@ -511,6 +514,7 @@ impl S3 for FS {
key: Some(v.name.to_owned()),
last_modified: v.mod_time.map(Timestamp::from),
size: Some(v.size as i64),
e_tag: v.etag.clone(),
..Default::default()
};
@@ -526,13 +530,23 @@ impl S3 for FS {
let key_count = objects.len() as i32;
let common_prefixes = object_infos
.prefixes
.into_iter()
.map(|v| CommonPrefix { prefix: Some(v) })
.collect();
let output = ListObjectsV2Output {
is_truncated: Some(object_infos.is_truncated),
continuation_token: object_infos.continuation_token,
next_continuation_token: object_infos.next_continuation_token,
key_count: Some(key_count),
max_keys: Some(key_count),
contents: Some(objects),
delimiter: Some(delimiter),
delimiter,
name: Some(bucket),
prefix: Some(prefix),
common_prefixes: Some(common_prefixes),
..Default::default()
};
@@ -542,9 +556,71 @@ impl S3 for FS {
async fn list_object_versions(
&self,
_req: S3Request<ListObjectVersionsInput>,
req: S3Request<ListObjectVersionsInput>,
) -> S3Result<S3Response<ListObjectVersionsOutput>> {
Err(s3_error!(NotImplemented, "ListObjectVersions is not implemented yet"))
let ListObjectVersionsInput {
bucket,
delimiter,
key_marker,
version_id_marker,
max_keys,
prefix,
..
} = req.input;
let prefix = prefix.unwrap_or_default();
let max_keys = max_keys.unwrap_or(1000);
let key_marker = key_marker.filter(|v| !v.is_empty());
let version_id_marker = version_id_marker.filter(|v| !v.is_empty());
let delimiter = delimiter.filter(|v| !v.is_empty());
let Some(store) = new_object_layer_fn() else {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let object_infos = store
.list_object_versions(&bucket, &prefix, key_marker, version_id_marker, delimiter.clone(), max_keys)
.await
.map_err(to_s3_error)?;
let objects: Vec<ObjectVersion> = object_infos
.objects
.iter()
.filter(|v| !v.name.is_empty())
.map(|v| {
ObjectVersion {
key: Some(v.name.to_owned()),
last_modified: v.mod_time.map(Timestamp::from),
size: Some(v.size as i64),
version_id: v.version_id.map(|v| v.to_string()),
is_latest: Some(v.is_latest),
e_tag: v.etag.clone(),
..Default::default() // TODO: another fields
}
})
.collect();
let key_count = objects.len() as i32;
let common_prefixes = object_infos
.prefixes
.into_iter()
.map(|v| CommonPrefix { prefix: Some(v) })
.collect();
let output = ListObjectVersionsOutput {
// is_truncated: Some(object_infos.is_truncated),
max_keys: Some(key_count),
delimiter,
name: Some(bucket),
prefix: Some(prefix),
common_prefixes: Some(common_prefixes),
versions: Some(objects),
..Default::default()
};
Ok(S3Response::new(output))
}
#[tracing::instrument(level = "debug", skip(self, req))]
rustfs/src/storage/error.rs
+4
View File
@@ -62,6 +62,10 @@ pub fn to_s3_error(err: Error) -> S3Error {
s3_error!(SlowDown, "Storage resources are insufficient for the write operation")
}
StorageError::DecommissionNotStarted => s3_error!(InvalidArgument, "Decommission Not Started"),
StorageError::VolumeNotFound(bucket) => {
s3_error!(NoSuchBucket, "bucket not found {}", bucket)
}
StorageError::InvalidPart(bucket, object, version_id) => {
s3_error!(
InvalidPart,