diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 3f51c1b66..98f2e0848 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -18,6 +18,7 @@ use crate::disk::error::{ 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}; @@ -452,7 +453,6 @@ impl LocalDisk { Ok(data) } - // FIXME: read_metadata only suport async fn read_metadata_with_dmtime(&self, file_path: impl AsRef) -> Result<(Vec, Option)> { check_path_length(file_path.as_ref().to_string_lossy().as_ref())?; @@ -471,18 +471,15 @@ impl LocalDisk { } let size = meta.len() as usize; - let mut bytes = Vec::new(); - bytes.try_reserve_exact(size)?; - // FIXME: real xl no data - 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, file_path: impl AsRef) -> Result> { @@ -1475,7 +1472,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 } @@ -1723,7 +1720,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)); } diff --git a/ecstore/src/file_meta.rs b/ecstore/src/file_meta.rs index 2a4fa0baf..3dae801d6 100644 --- a/ecstore/src/file_meta.rs +++ b/ecstore/src/file_meta.rs @@ -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, @@ -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 { @@ -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 { 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); @@ -1994,9 +2010,96 @@ async fn get_file_info(buf: &[u8], volume: &str, path: &str, version_id: &str, o Ok(fi) } + +async fn read_more( + reader: &mut R, + buf: &mut Vec, + 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(reader: &mut R, size: usize) -> Result> { + 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 { + use std::fs; + use std::fs::File; + use std::os::unix::fs::MetadataExt; + use super::*; #[test] @@ -2012,15 +2115,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 +2206,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) +} diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 2dbdebad6..ddd417b6e 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -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,23 @@ 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 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; +mod store_list_objects; +pub mod utils; pub mod xhttp; pub use global::new_object_layer_fn; diff --git a/ecstore/src/notification_sys.rs b/ecstore/src/notification_sys.rs index 76df1e637..8a994d939 100644 --- a/ecstore/src/notification_sys.rs +++ b/ecstore/src/notification_sys.rs @@ -46,6 +46,11 @@ pub struct NotificationPeerErr { } impl NotificationSys { + pub fn rest_client_from_hash(&self, s:&str) ->Option{ + + + None + } pub async fn delete_policy(&self) -> Vec { unimplemented!() } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index a09cc7f3c..e583545d2 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -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 { - // 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 { + // // 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> { - 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> { + // 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(); @@ -1520,29 +1518,32 @@ impl StorageAPI for ECStore { continuation_token: &str, delimiter: &str, max_keys: i32, - _fetch_owner: bool, - _start_after: &str, + fetch_owner: bool, + start_after: &str, ) -> Result { - let opts = ListPathOptions { - bucket: bucket.to_string(), - limit: max_keys, - prefix: prefix.to_owned(), - ..Default::default() - }; + self.inner_list_objects_v2(&bucket, &prefix, &continuation_token, &delimiter, max_keys, fetch_owner, start_after) + .await - let info = self.list_path(&opts, delimiter).await?; + // let opts = ListPathOptions { + // bucket: bucket.to_string(), + // limit: max_keys, + // prefix: prefix.to_owned(), + // ..Default::default() + // }; - // warn!("list_objects_v2 info {:?}", info); + // let info = self.list_path(&opts, delimiter).await?; - 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, - }; + // // warn!("list_objects_v2 info {:?}", info); - Ok(v2) + // 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) } async fn list_object_versions( &self, @@ -2238,7 +2239,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: &str) -> Result<()> { if !is_meta_bucketname(bucket) && check_valid_bucket_name_strict(bucket).is_err() { return Err(Error::new(StorageError::BucketNameInvalid(bucket.to_string()))); } diff --git a/ecstore/src/list_objects.rs b/ecstore/src/store_list_objects.rs similarity index 96% rename from ecstore/src/list_objects.rs rename to ecstore/src/store_list_objects.rs index bb1f4cd2d..119cb738f 100644 --- a/ecstore/src/list_objects.rs +++ b/ecstore/src/store_list_objects.rs @@ -101,6 +101,7 @@ impl ListPathOptions { } impl ECStore { + #[allow(clippy::too_many_arguments)] pub async fn inner_list_objects_v2( &self, bucket: &str, @@ -120,7 +121,7 @@ impl ECStore { }; self.list_objects_generic(bucket, prefix, marker, delimiter, max_keys).await?; - + // FIXME:TODO: unimplemented!() } @@ -145,6 +146,8 @@ impl ECStore { let merged = self.list_path(&opts).await?; + // FIXME:TODO: + todo!() } @@ -159,10 +162,8 @@ impl ECStore { 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.marker.is_empty() && !o.prefix.is_empty() && !o.marker.starts_with(&o.prefix) { + return Err(Error::new(std::io::Error::from(ErrorKind::UnexpectedEof))); } if o.limit == 0 { @@ -194,6 +195,7 @@ impl ECStore { o.create = false; } + // FIXME:TODO: todo!() // let mut opts = opts.clone();