From edb1e692f136d284af8772b153f1c82a39b7d569 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Thu, 24 Oct 2024 10:16:52 +0800 Subject: [PATCH] bitrot(1) Signed-off-by: junxiang Mu <1948535941@qq.com> --- Cargo.lock | 17 ++++ ecstore/Cargo.toml | 2 + ecstore/src/bitrot.rs | 167 +++++++++++++++++++++++++++++++++++++ ecstore/src/disk/local.rs | 61 +++++++++++--- ecstore/src/disk/mod.rs | 5 ++ ecstore/src/disk/remote.rs | 9 +- ecstore/src/lib.rs | 1 + ecstore/src/store_api.rs | 15 +++- ecstore/src/utils/path.rs | 12 +++ 9 files changed, 274 insertions(+), 15 deletions(-) create mode 100644 ecstore/src/bitrot.rs diff --git a/Cargo.lock b/Cargo.lock index 4a670527a..a977e6def 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -276,6 +276,15 @@ version = "2.6.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b048fb63fd8b5923fc5aa7b340d8e156aec7ec02f0c78fa8a6ddc2613f6f71de" +[[package]] +name = "blake2" +version = "0.10.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "46502ad458c9a52b69d4d4d32775c788b7a1b85e8bc9d482d92250fc0e3f8efe" +dependencies = [ + "digest", +] + [[package]] name = "block-buffer" version = "0.10.4" @@ -479,6 +488,7 @@ dependencies = [ "async-trait", "backon", "base64-simd", + "blake2", "byteorder", "bytes", "common", @@ -486,6 +496,7 @@ dependencies = [ "futures", "glob", "hex-simd", + "highway", "http", "lazy_static", "lock", @@ -796,6 +807,12 @@ dependencies = [ "vsimd", ] +[[package]] +name = "highway" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c706f1711006204c2ba8fb1a7bd55f689bbf7feca9ff40325206b5e140cff6df" + [[package]] name = "hmac" version = "0.12.1" diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index b66875f03..523ef2994 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -11,6 +11,7 @@ rust-version.workspace = true [dependencies] async-trait.workspace = true backon.workspace = true +blake2 = "0.10.6" bytes.workspace = true common.workspace = true glob = "0.3.1" @@ -23,6 +24,7 @@ serde_json.workspace = true tracing-error.workspace = true s3s.workspace = true http.workspace = true +highway = "1.2.0" url.workspace = true uuid = { version = "1.8.0", features = ["v4", "fast-rng", "serde"] } reed-solomon-erasure = { version = "6.0.0", features = ["simd-accel"] } diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs new file mode 100644 index 000000000..822eb877d --- /dev/null +++ b/ecstore/src/bitrot.rs @@ -0,0 +1,167 @@ +use std::collections::HashMap; + +use blake2::{Blake2b, Blake2b512}; +use highway::{HighwayHash, HighwayHasher, Key}; +use lazy_static::lazy_static; +use sha2::{Digest, Sha256}; + +use crate::{ + disk::DiskStore, + erasure::{ReadAt, Write}, + error::Result, + store_api::BitrotAlgorithm, +}; + +lazy_static! { + static ref BITROT_ALGORITHMS: HashMap = { + let mut m = HashMap::new(); + m.insert(BitrotAlgorithm::SHA256, "sha256"); + m.insert(BitrotAlgorithm::BLAKE2b512, "blake2b"); + m.insert(BitrotAlgorithm::HighwayHash256, "highwayhash256"); + m.insert(BitrotAlgorithm::HighwayHash256S, "highwayhash256S"); + m + }; +} + +// const MAGIC_HIGHWAY_HASH256_KEY: &[u8] = &[ +// 0x4b, 0xe7, 0x34, 0xfa, 0x8e, 0x23, 0x8a, 0xcd, 0x26, 0x3e, 0x83, 0xe6, 0xbb, 0x96, 0x85, 0x52, 0x04, 0x0f, 0x93, 0x5d, 0xa3, +// 0x9f, 0x44, 0x14, 0x97, 0xe0, 0x9d, 0x13, 0x22, 0xde, 0x36, 0xa0, +// ]; +const MAGIC_HIGHWAY_HASH256_KEY: &[u64; 4] = &[3, 4, 2, 1]; + +pub enum Hasher { + SHA256(Sha256), + HighwayHash256(HighwayHasher), + BLAKE2b512(Blake2b512), +} + +impl Hasher { + pub fn update(&mut self, data: impl AsRef<[u8]>) { + match self { + Hasher::SHA256(core_wrapper) => { + core_wrapper.update(data); + } + Hasher::HighwayHash256(highway_hasher) => { + highway_hasher.append(data.as_ref()); + } + Hasher::BLAKE2b512(core_wrapper) => { + core_wrapper.update(data); + } + } + } + + pub fn finalize(self) -> Vec { + match self { + Hasher::SHA256(core_wrapper) => core_wrapper.finalize().to_vec(), + Hasher::HighwayHash256(highway_hasher) => highway_hasher + .finalize256() + .iter() + .flat_map(|&n| n.to_le_bytes()) // 使用小端字节序转换 + .collect(), + Hasher::BLAKE2b512(core_wrapper) => core_wrapper.finalize().to_vec(), + } + } +} + +impl BitrotAlgorithm { + pub fn new(&self) -> Hasher { + match self { + BitrotAlgorithm::SHA256 => Hasher::SHA256(Sha256::new()), + BitrotAlgorithm::HighwayHash256 | BitrotAlgorithm::HighwayHash256S => { + let key = Key(*MAGIC_HIGHWAY_HASH256_KEY); + Hasher::HighwayHash256(HighwayHasher::new(key)) + } + BitrotAlgorithm::BLAKE2b512 => Hasher::BLAKE2b512(Blake2b512::new()), + } + } + + pub fn available(&self) -> bool { + BITROT_ALGORITHMS.get(self).is_some() + } + + pub fn string(&self) -> String { + BITROT_ALGORITHMS.get(self).map_or("".to_string(), |s| s.to_string()) + } +} + +pub struct BitrotVerifier { + algorithm: BitrotAlgorithm, + sum: Vec, +} + +impl BitrotVerifier { + pub fn new(algorithm: BitrotAlgorithm, checksum: &[u8]) -> BitrotVerifier { + BitrotVerifier { + algorithm, + sum: checksum.to_vec(), + } + } +} + +pub fn bitrot_algorithm_from_string(s: &str) -> BitrotAlgorithm { + for (k, v) in BITROT_ALGORITHMS.iter() { + if *v == s { + return k.clone(); + } + } + + BitrotAlgorithm::HighwayHash256S +} + +type BitrotWriter = Box; + +pub fn new_bitrot_writer( + disk: DiskStore, + orig_volume: &str, + volume: &str, + file_path: &str, + length: usize, + algo: BitrotAlgorithm, + shard_size: usize, +) -> BitrotWriter { + todo!() +} + +pub struct WholeBitrotWriter { + disk: DiskStore, + volume: String, + file_path: String, + shard_size: usize, + hash: Hasher, +} + +impl WholeBitrotWriter { + pub fn new(disk: DiskStore, volume: &str, file_path: &str, algo: BitrotAlgorithm, shard_size: usize) -> Self { + WholeBitrotWriter { + disk, + volume: volume.to_string(), + file_path: file_path.to_string(), + shard_size, + hash: algo.new(), + } + } +} + +#[async_trait::async_trait] +impl Write for WholeBitrotWriter { + async fn write(&mut self, buf: &[u8]) -> Result<()> { + let mut file = self.disk.append_file(&self.volume, &self.file_path).await?; + let _ = file.write(buf).await?; + self.hash.update(buf); + + Ok(()) + } +} + +pub struct WholeBitrotReader { + disk: DiskStore, + volume: String, + file_path: String, +} + +#[async_trait::async_trait] +impl ReadAt for WholeBitrotReader { + async fn read_at(&mut self, offset: usize, length: usize) -> Result<(Vec, usize)> { + todo!() + } +} diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 272b87276..ae77cef42 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -2,9 +2,9 @@ use super::error::{is_sys_err_io, is_sys_err_not_empty, is_sys_err_too_many_file use super::os::is_root_disk; use super::{endpoint::Endpoint, error::DiskError, format::FormatV3}; use super::{ - os, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter, - Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, - WalkDirOptions, + os, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics, FileInfoVersions, + FileReader, FileWriter, Info, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, + UpdateMetadataOpts, VolumeInfo, WalkDirOptions, }; use crate::cache_value::cache::{Cache, Opts}; use crate::disk::error::{ @@ -15,16 +15,16 @@ 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::store_api::BitrotAlgorithm; use crate::utils::fs::{lstat, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY}; -use crate::utils::path::{clean, has_suffix, SLASH_SEPARATOR}; use crate::utils::os::get_info; +use crate::utils::path::{clean, has_suffix, SLASH_SEPARATOR}; use crate::{ file_meta::FileMeta, store_api::{FileInfo, RawFileInfo}, utils, }; use path_absolutize::Absolutize; -use tokio::runtime::Runtime; use std::fmt::Debug; use std::sync::atomic::{AtomicU32, Ordering}; use std::sync::Arc; @@ -36,6 +36,7 @@ use std::{ use time::OffsetDateTime; use tokio::fs::{self, File}; use tokio::io::{AsyncReadExt, AsyncWriteExt, ErrorKind}; +use tokio::runtime::Runtime; use tokio::sync::RwLock; use tracing::{error, warn}; use uuid::Uuid; @@ -149,19 +150,17 @@ impl LocalDisk { if root { return Err(Error::new(DiskError::DriveIsRoot)); } - + // disk_info.healing = Ok(disk_info) - }, - Err(err) => { - Err(err) } + Err(err) => Err(err), } }) }; - + let cache = Cache::new(Box::new(update_fn), Duration::from_secs(1), Opts::default()); - + // TODO: DIRECT suport // TODD: DiskInfo let mut disk = Self { @@ -650,6 +649,22 @@ impl LocalDisk { fn get_metrics(&self) -> DiskMetrics { DiskMetrics::default() } + + async fn bitrot_verify( + &self, + part_path: &PathBuf, + part_size: u64, + algo: BitrotAlgorithm, + sum: &[u8], + shard_size: u64, + ) -> Result<()> { + let file = utils::fs::open_file(part_path, O_CREATE | O_WRONLY) + .await + .map_err(os_err_to_file_err)?; + + + todo!() + } } fn is_root_path(path: impl AsRef) -> bool { @@ -846,6 +861,30 @@ impl DiskAPI for LocalDisk { Ok(()) } + async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result { + let volume_dir = self.get_bucket_path(volume)?; + if !skip_access_checks(volume) { + if let Err(e) = utils::fs::access(&volume_dir).await { + return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); + } + } + + let mut resp = CheckPartsResp { + results: Vec::with_capacity(fi.parts.len()), + }; + + let erasure = fi.erasure; + fi.parts.iter().enumerate().for_each(|(i, part)| { + 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)); + + self.bi + }); + } + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { let src_volume_dir = self.get_bucket_path(src_volume)?; let dst_volume_dir = self.get_bucket_path(dst_volume)?; diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 0fa68ff2f..89a30746b 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -127,6 +127,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { // 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; // StatInfoFile // ReadParts async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; @@ -136,6 +137,10 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn disk_info(&self, opts: &DiskInfoOptions) -> Result; } +pub struct CheckPartsResp { + pub results: Vec, +} + #[derive(Debug, Serialize, Deserialize)] pub struct UpdateMetadataOpts { pub no_persistence: bool, diff --git a/ecstore/src/disk/remote.rs b/ecstore/src/disk/remote.rs index 4b82c43cd..8d120a065 100644 --- a/ecstore/src/disk/remote.rs +++ b/ecstore/src/disk/remote.rs @@ -15,9 +15,7 @@ use tracing::info; use uuid::Uuid; use super::{ - endpoint::Endpoint, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, - FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, - RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, + endpoint::Endpoint, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader, FileWriter, MetaCacheEntry, ReadMultipleReq, ReadMultipleResp, ReadOptions, RemoteFileReader, RemoteFileWriter, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions }; use crate::{ disk::error::DiskError, @@ -163,6 +161,11 @@ impl DiskAPI for RemoteDisk { Ok(()) } + + async fn verify_file(&self, volume: &str, path: &str, fi: FileInfo) -> Result { + unimplemented!() + } + async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { info!("rename_part"); let mut client = node_service_time_out_client(&self.addr) diff --git a/ecstore/src/lib.rs b/ecstore/src/lib.rs index 5d1346b25..4533f1266 100644 --- a/ecstore/src/lib.rs +++ b/ecstore/src/lib.rs @@ -1,3 +1,4 @@ +pub mod bitrot; pub mod cache_value; mod chunk_stream; mod config; diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 1df9741d6..51005a071 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -12,6 +12,7 @@ use http::HeaderMap; use rmp_serde::Serializer; use s3s::dto::StreamingBlob; use serde::{Deserialize, Serialize}; +use sha2::Sha256; use time::OffsetDateTime; use uuid::Uuid; @@ -240,6 +241,18 @@ pub struct ErasureInfo { pub checksums: Vec, } +impl ErasureInfo { + pub fn get_checksum_info(&self, part_number: usize) -> ChecksumInfo { + for sum in &self.checksums { + if sum.part_number == part_number { + return sum.clone(); + } + } + + ChecksumInfo {algorithm: BitrotAlgorithm::HighwayHash256S, ..Default::default()} + } +} + #[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)] // ChecksumInfo - carries checksums of individual scattered parts per disk. pub struct ChecksumInfo { @@ -248,7 +261,7 @@ pub struct ChecksumInfo { pub hash: Vec, } -#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone)] +#[derive(Serialize, Deserialize, Debug, PartialEq, Default, Clone, Eq, Hash)] // BitrotAlgorithm specifies a algorithm used for bitrot protection. pub enum BitrotAlgorithm { // SHA256 represents the SHA-256 hash function diff --git a/ecstore/src/utils/path.rs b/ecstore/src/utils/path.rs index bf33313c0..fc35031fc 100644 --- a/ecstore/src/utils/path.rs +++ b/ecstore/src/utils/path.rs @@ -1,3 +1,5 @@ +use std::path::PathBuf; + const GLOBAL_DIR_SUFFIX: &str = "__XLDIR__"; pub const SLASH_SEPARATOR: &str = "/"; @@ -50,6 +52,16 @@ pub fn has_profix(s: &str, prefix: &str) -> bool { s.starts_with(prefix) } +pub fn path_join(elem: &[PathBuf]) -> PathBuf { + let mut joined_path = PathBuf::new(); + + for path in elem { + joined_path.push(path); + } + + joined_path +} + pub struct LazyBuf { s: String, buf: Option>,