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

heal object(1)

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2024-11-01 17:38:50 +08:00
parent 640cf1caf3
commit a8d36fba8f
9 changed files with 574 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/bitrot.rs
+41 -15
View File
@@ -1,4 +1,8 @@
use std::{any::Any, collections::HashMap, io::Cursor};
use std::{
any::Any,
collections::HashMap,
io::{Cursor, Read},
};
use blake2::Blake2b512;
use highway::{HighwayHash, HighwayHasher, Key};
@@ -205,21 +209,49 @@ pub fn bitrot_shard_file_size(size: usize, shard_size: usize, algo: BitrotAlgori
}
pub fn bitrot_verify(
r: Cursor<Vec<u8>>,
_want_size: usize,
_part_size: usize,
r: &mut Cursor<Vec<u8>>,
want_size: usize,
part_size: usize,
algo: BitrotAlgorithm,
want: Vec<u8>,
_shard_size: usize,
mut shard_size: usize,
) -> Result<()> {
if algo != BitrotAlgorithm::HighwayHash256S {
let mut h = algo.new();
h.update(r.into_inner());
h.update(r.get_ref());
if h.finalize() != want {
return Err(Error::new(DiskError::FileCorrupt));
}
return Ok(());
}
todo!()
let mut h = algo.new();
let mut hash_buf = vec![0; h.size()];
let mut left = want_size;
if left != bitrot_shard_file_size(part_size, shard_size, algo) {
return Err(Error::new(DiskError::FileCorrupt));
}
while left > 0 {
h.reset();
let n = r.read(&mut hash_buf)?;
left -= n;
if left < shard_size {
shard_size = left;
}
let mut buf = vec![0; shard_size];
let read = r.read(&mut buf)?;
h.update(buf);
left -= read;
if h.clone().finalize() != hash_buf[0..n] {
return Err(Error::new(DiskError::FileCorrupt));
}
}
Ok(())
}
pub struct WholeBitrotWriter {
@@ -295,7 +327,7 @@ impl ReadAt for WholeBitrotReader {
return Err(Error::new(DiskError::LessData));
}
return Ok((buf.split_off(length), length));
return Ok((buf.drain(0..length).collect::<Vec<_>>(), length));
}
Err(Error::new(DiskError::LessData))
@@ -453,10 +485,7 @@ mod test {
use tempfile::TempDir;
use crate::{
bitrot::{new_bitrot_writer, BITROT_ALGORITHMS},
disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption},
error::{Error, Result},
store_api::BitrotAlgorithm,
bitrot::{new_bitrot_writer, BITROT_ALGORITHMS}, disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption}, error::{Error, Result}, store_api::BitrotAlgorithm
};
use super::{bitrot_writer_sum, new_bitrot_reader};
@@ -512,9 +541,6 @@ mod test {
#[tokio::test]
async fn test_all_bitrot_algorithms() -> Result<()> {
for algo in BITROT_ALGORITHMS.keys() {
if *algo != BitrotAlgorithm::HighwayHash256S {
continue;
}
test_bitrot_reader_writer_algo(algo.clone()).await?;
}
ecstore/src/disk/error.rs
+8
View File
@@ -100,6 +100,12 @@ pub enum DiskError {
#[error("more data was sent than what was advertised")]
MoreData,
#[error("outdated XL meta")]
OutdatedXLMeta,
#[error("part missing or corrupt")]
PartMissingOrCorrupt,
}
impl DiskError {
@@ -202,6 +208,8 @@ pub fn clone_disk_err(e: &DiskError) -> Error {
DiskError::CrossDeviceLink => Error::new(DiskError::CrossDeviceLink),
DiskError::LessData => Error::new(DiskError::LessData),
DiskError::MoreData => Error::new(DiskError::MoreData),
DiskError::OutdatedXLMeta => Error::new(DiskError::OutdatedXLMeta),
DiskError::PartMissingOrCorrupt => Error::new(DiskError::PartMissingOrCorrupt),
}
}
ecstore/src/disk/local.rs
+94 -14
View File
@@ -6,6 +6,7 @@ use super::{
FileReader, FileWriter, Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
};
use crate::bitrot::bitrot_verify;
use crate::cache_value::cache::{Cache, Opts};
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,
@@ -15,8 +16,9 @@ use crate::disk::os::check_path_length;
use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
use crate::error::{Error, Result};
use crate::global::{GLOBAL_IsErasureSD, GLOBAL_RootDiskThreshold};
use crate::set_disk::{conv_part_err_to_int, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN, CHECK_PART_VOLUME_NOT_FOUND};
use crate::store_api::BitrotAlgorithm;
use crate::utils::fs::{lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY};
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, SLASH_SEPARATOR};
use crate::{
@@ -26,6 +28,8 @@ use crate::{
};
use path_absolutize::Absolutize;
use std::fmt::Debug;
use std::io::Cursor;
use std::os::unix::fs::MetadataExt;
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -38,7 +42,7 @@ use tokio::fs::{self, File};
use tokio::io::{AsyncReadExt, AsyncWriteExt, ErrorKind};
use tokio::runtime::Runtime;
use tokio::sync::RwLock;
use tracing::{error, warn};
use tracing::{error, info, warn};
use uuid::Uuid;
#[derive(Debug)]
@@ -180,7 +184,7 @@ impl LocalDisk {
// format_data: Mutex::new(format_data),
// format_last_check: Mutex::new(format_last_check),
};
let (info, root) = get_disk_info(root_clone).await?;
let (info, _root) = get_disk_info(root_clone).await?;
disk.major = info.major;
disk.minor = info.minor;
disk.fstype = info.fstype;
@@ -653,17 +657,18 @@ impl LocalDisk {
async fn bitrot_verify(
&self,
part_path: &PathBuf,
part_size: u64,
part_size: usize,
algo: BitrotAlgorithm,
sum: &[u8],
shard_size: u64,
shard_size: usize,
) -> Result<()> {
let file = utils::fs::open_file(part_path, O_CREATE | O_WRONLY)
let mut file = utils::fs::open_file(part_path, O_CREATE | O_WRONLY)
.await
.map_err(os_err_to_file_err)?;
todo!()
let mut data = Vec::new();
let n = file.read_to_end(&mut data).await?;
bitrot_verify(&mut Cursor::new(data), n, part_size, algo, sum.to_vec(), shard_size)
}
}
@@ -861,7 +866,7 @@ impl DiskAPI for LocalDisk {
Ok(())
}
async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result<CheckPartsResp> {
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
let volume_dir = self.get_bucket_path(volume)?;
if !skip_access_checks(volume) {
if let Err(e) = utils::fs::access(&volume_dir).await {
@@ -873,17 +878,92 @@ impl DiskAPI for LocalDisk {
results: Vec::with_capacity(fi.parts.len()),
};
let erasure = fi.erasure;
fi.parts.iter().enumerate().for_each(|(i, part)| {
let erasure = &fi.erasure;
for (i, part) in fi.parts.iter().enumerate() {
let checksum_info = erasure.get_checksum_info(part.number);
let part_path = Path::new(&volume_dir)
.join(path)
.join(fi.data_dir.map_or("".to_string(), |dir| dir.to_string()))
.join(format!("part.{}", part.number));
let err = match self
.bitrot_verify(
&part_path,
erasure.shard_file_size(part.size),
checksum_info.algorithm,
&checksum_info.hash,
erasure.shard_size(erasure.block_size),
)
.await
{
Ok(_) => None,
Err(err) => Some(err),
};
resp.results[i] = conv_part_err_to_int(&err);
if resp.results[i] == CHECK_PART_UNKNOWN {
if let Some(err) = err {
match err.downcast_ref::<DiskError>() {
Some(DiskError::FileAccessDenied) => {}
_ => {
info!("part unknown, disk: {}, path: {:?}", self.to_string(), part_path);
}
}
}
}
}
// self.bi
});
todo!()
Ok(resp)
}
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
let volume_dir = self.get_bucket_path(&volume)?;
check_path_length(volume_dir.join(path).to_string_lossy().as_ref())?;
let mut resp = CheckPartsResp {
results: vec![0; fi.parts.len()],
};
for (i, part) in fi.parts.iter().enumerate() {
let file_path = Path::new(&volume_dir)
.join(path)
.join(fi.data_dir.map_or("".to_string(), |dir| dir.to_string()))
.join(format!("part.{}", part.number));
match lstat(file_path).await {
Ok(st) => {
if st.is_dir() {
resp.results[i] = CHECK_PART_FILE_NOT_FOUND;
continue;
}
if (st.size() as usize) < fi.erasure.shard_file_size(part.size) {
resp.results[i] = CHECK_PART_FILE_CORRUPT;
continue;
}
resp.results[i] = CHECK_PART_SUCCESS;
},
Err(err) => {
match os_err_to_file_err(err).downcast_ref() {
Some(DiskError::FileNotFound) => {
if !skip_access_checks(volume) {
if let Err(err) = access(&volume_dir).await {
match err.kind() {
ErrorKind::NotFound => {
resp.results[i] = CHECK_PART_VOLUME_NOT_FOUND;
continue;
},
_ => {},
}
}
}
resp.results[i] = CHECK_PART_FILE_NOT_FOUND;
},
_ => {},
}
continue;
}
}
}
Ok(resp)
}
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
ecstore/src/disk/mod.rs
+4 -2
View File
@@ -124,10 +124,11 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
// ReadFileStream
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>;
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()>;
// CheckParts
async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>;
// VerifyFile
async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result<CheckPartsResp>;
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp>;
// CheckParts
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp>;
// StatInfoFile
// ReadParts
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
@@ -137,6 +138,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo>;
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct CheckPartsResp {
pub results: Vec<usize>,
}
ecstore/src/disk/remote.rs
+47 -6
View File
@@ -4,10 +4,7 @@ use futures::lock::Mutex;
use protos::{
node_service_time_out_client,
proto_gen::node_service::{
DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest,
ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest,
ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest, UpdateMetadataRequest,
WalkDirRequest, WriteAllRequest, WriteMetadataRequest,
CheckPartsRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest, MakeVolumesRequest, ReadAllRequest, ReadMultipleRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequst, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WalkDirRequest, WriteAllRequest, WriteMetadataRequest
},
};
use tonic::Request;
@@ -162,8 +159,52 @@ impl DiskAPI for RemoteDisk {
Ok(())
}
async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result<CheckPartsResp> {
unimplemented!()
async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
info!("verify_file");
let file_info = serde_json::to_string(&fi)?;
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(VerifyFileRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
path: path.to_string(),
file_info,
});
let response = client.verify_file(request).await?.into_inner();
if !response.success {
return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
}
let check_parts_resp = serde_json::from_str::<CheckPartsResp>(&response.check_parts_resp)?;
Ok(check_parts_resp)
}
async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result<CheckPartsResp> {
info!("check_parts");
let file_info = serde_json::to_string(&fi)?;
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(CheckPartsRequest {
disk: self.root.to_string_lossy().to_string(),
volume: volume.to_string(),
path: path.to_string(),
file_info,
});
let response = client.check_parts(request).await?.into_inner();
if !response.success {
return Err(Error::from_string(response.error_info.unwrap_or("".to_string())));
}
let check_parts_resp = serde_json::from_str::<CheckPartsResp>(&response.check_parts_resp)?;
Ok(check_parts_resp)
}
async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec<u8>) -> Result<()> {
ecstore/src/store_api.rs
+65 -3
View File
@@ -1,11 +1,10 @@
use std::collections::HashMap;
use crate::{
error::{Error, Result},
heal::{
disk::error::DiskError, error::{Error, Result}, heal::{
heal_commands::{HealOpts, HealResultItem},
heal_ops::HealObjectFn,
},
}
};
use futures::StreamExt;
use http::HeaderMap;
@@ -17,6 +16,8 @@ use uuid::Uuid;
pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
pub const BLOCK_SIZE_V2: usize = 1048576; // 1M
pub const RESERVED_METADATA_PREFIX: &str = "X-Rustfs-Internal-";
pub const RESERVED_METADATA_PREFIX_LOWER: &str = "X-Rustfs-Internal-";
// #[derive(Debug, Clone)]
#[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
@@ -172,6 +173,7 @@ impl FileInfo {
parts: self.parts.clone(),
is_latest: self.is_latest,
tags: self.tags.clone(),
..Default::default()
}
}
// to_part_offset 取offset 所在的part index, 返回part index, offset
@@ -250,6 +252,29 @@ impl ErasureInfo {
ChecksumInfo {algorithm: DEFAULT_BITROT_ALGO, ..Default::default()}
}
// 算出每个分片大小
pub fn shard_size(&self, data_size: usize) -> usize {
(data_size + self.data_blocks - 1) / self.data_blocks
}
// returns final erasure size from original size.
pub fn shard_file_size(&self, total_size: usize) -> usize {
if total_size == 0 {
return 0;
}
let num_shards = total_size / self.block_size;
let last_block_size = total_size % self.block_size;
let last_shard_size = (last_block_size + self.data_blocks - 1) / self.data_blocks;
num_shards * self.shard_size(self.block_size) + last_shard_size
// // 因为写入的时候ec需要补全，所以最后一个长度应该也是一样的
// if last_block_size != 0 {
// num_shards += 1
// }
// num_shards * self.shard_size(self.block_size)
}
}
#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)]
@@ -481,7 +506,10 @@ pub struct ObjectInfo {
pub name: String,
pub mod_time: Option<OffsetDateTime>,
pub size: usize,
// Actual size is the real size of the object uploaded by client.
pub actual_size: Option<usize>,
pub is_dir: bool,
pub user_defined: HashMap<String, String>,
pub parity_blocks: usize,
pub data_blocks: usize,
pub version_id: Option<Uuid>,
@@ -491,6 +519,40 @@ pub struct ObjectInfo {
pub tags: Option<HashMap<String, String>>,
}
impl ObjectInfo {
pub fn is_compressed(&self) -> bool {
self.user_defined.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX))
}
pub fn get_actual_size(&self) -> Result<usize> {
if let Some(actual_size) = self.actual_size {
return Ok(actual_size);
}
if self.is_compressed() {
if let Some(size_str) = self.user_defined.get(&format!("{}actual-size", RESERVED_METADATA_PREFIX)) {
if !size_str.is_empty() {
// Todo: deal with error
let size = size_str.parse::<usize>()?;
return Ok(size);
}
}
let mut actual_size = 0;
self.parts.iter().for_each(|part| {
actual_size += part.actual_size;
});
if actual_size == 0 && actual_size != self.size {
return Err(Error::from_string("invalid decompressed size"));
}
return Ok(actual_size);
}
// TODO: IsEncrypted
Ok(self.size)
}
}
#[derive(Debug, Default)]
pub struct ListObjectsInfo {
// Indicates whether the returned list objects response is truncated. A