From e6cd184cd60b04a691d9d70befebe9fba3997593 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Wed, 30 Oct 2024 11:26:49 +0800 Subject: [PATCH] bitrot test(1) Signed-off-by: junxiang Mu <1948535941@qq.com> --- Cargo.lock | 1 + Cargo.toml | 1 + ecstore/Cargo.toml | 1 + ecstore/src/bitrot.rs | 157 +++++++++++++++++++++++--------- ecstore/src/disk/mod.rs | 14 ++- ecstore/src/erasure.rs | 2 + ecstore/src/utils/os/linux.rs | 162 ++++++++++++++++++++-------------- 7 files changed, 230 insertions(+), 108 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a977e6def..2c884b2b6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -517,6 +517,7 @@ dependencies = [ "serde_json", "sha2", "siphasher", + "tempfile", "thiserror", "time", "tokio", diff --git a/Cargo.toml b/Cargo.toml index 2a50af2f9..c81b87305 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -53,6 +53,7 @@ s3s = { git = "https://github.com/Nugine/s3s.git", rev = "0caf79822ae1f2e7a7fad3 s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "0caf79822ae1f2e7a7fad3c4a093c94dd2e2c33f" } serde = { version = "1.0.210", features = ["derive"] } serde_json = "1.0.128" +tempfile = "3.13.0" thiserror = "1.0.64" time = { version = "0.3.36", features = [ "std", diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index 523ef2994..a7ac29104 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -43,6 +43,7 @@ base64-simd = "0.8.0" sha2 = { version = "0.10.8", features = ["asm"] } hex-simd = "0.8.0" path-clean = "1.0.1" +tempfile.workspace = true tokio = { workspace = true, features = ["io-util", "sync"] } tokio-stream = "0.1.15" tonic.workspace = true diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index f658fd095..dd6ab8c01 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -1,7 +1,6 @@ use std::{any::Any, collections::HashMap, io::Cursor}; use blake2::Blake2b512; -use hex_simd::{decode_to_vec, encode_to_string}; use highway::{HighwayHash, HighwayHasher, Key}; use lazy_static::lazy_static; use sha2::{digest::core_api::BlockSizeUser, Digest, Sha256}; @@ -30,6 +29,7 @@ lazy_static! { // ]; const MAGIC_HIGHWAY_HASH256_KEY: &[u64; 4] = &[3, 4, 2, 1]; +#[derive(Clone)] pub enum Hasher { SHA256(Sha256), HighwayHash256(HighwayHasher), @@ -85,8 +85,8 @@ impl Hasher { Hasher::HighwayHash256(highway_hasher) => { let key = Key(*MAGIC_HIGHWAY_HASH256_KEY); *highway_hasher = HighwayHasher::new(key); - } , - Hasher::BLAKE2b512(core_wrapper) =>core_wrapper.reset(), + } + Hasher::BLAKE2b512(core_wrapper) => core_wrapper.reset(), } } } @@ -165,6 +165,14 @@ pub fn new_bitrot_reader( Box::new(WholeBitrotReader::new(disk, bucket, file_path, algo, till_offset, sum)) } +pub fn bitrot_writer_sum(w: &BitrotWriter) -> Vec { + if let Some(w) = w.as_any().downcast_ref::() { + return w.hash.clone().finalize(); + } + + Vec::new() +} + pub fn bitrot_shard_file_size(size: i64, _shard_size: i64, algo: BitrotAlgorithm) -> i64 { if algo != BitrotAlgorithm::HighwayHash256S { return size; @@ -195,21 +203,17 @@ pub struct WholeBitrotWriter { disk: DiskStore, volume: String, file_path: String, - shard_size: usize, - hash: Hasher, + _shard_size: usize, + pub hash: Hasher, } impl WholeBitrotWriter { - fn as_any(&self) -> &dyn Any { - self - } - 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, + _shard_size: shard_size, hash: algo.new(), } } @@ -217,6 +221,10 @@ impl WholeBitrotWriter { #[async_trait::async_trait] impl Write for WholeBitrotWriter { + fn as_any(&self) -> &dyn Any { + self + } + 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?; @@ -230,7 +238,7 @@ pub struct WholeBitrotReader { disk: DiskStore, volume: String, file_path: String, - verifier: BitrotVerifier, + _verifier: BitrotVerifier, till_offset: usize, buf: Option>, } @@ -241,7 +249,7 @@ impl WholeBitrotReader { disk, volume: volume.to_string(), file_path: file_path.to_string(), - verifier: BitrotVerifier::new(algo, sum), + _verifier: BitrotVerifier::new(algo, sum), till_offset, buf: None, } @@ -270,39 +278,108 @@ impl ReadAt for WholeBitrotReader { } } -#[test] -fn bitrot_self_test() -> Result<()> { - let mut checksums = HashMap::new(); - checksums.insert(BitrotAlgorithm::SHA256, "a7677ff19e0182e4d52e3a3db727804abc82a5818749336369552e54b838b004"); - checksums.insert(BitrotAlgorithm::BLAKE2b512, "e519b7d84b1c3c917985f544773a35cf265dcab10948be3550320d156bab612124a5ae2ae5a8c73c0eea360f68b0e28136f26e858756dbfe7375a7389f26c669"); - checksums.insert(BitrotAlgorithm::HighwayHash256, "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b"); - checksums.insert(BitrotAlgorithm::HighwayHash256S, "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b"); +#[cfg(test)] +mod test { + use std::{collections::HashMap, fs}; - let iter = [BitrotAlgorithm::SHA256, BitrotAlgorithm::BLAKE2b512, BitrotAlgorithm:: HighwayHash256]; + use hex_simd::decode_to_vec; + use tempfile::TempDir; - for algo in iter.iter() { - if !algo.available() || *algo != BitrotAlgorithm::HighwayHash256 { - continue; - } - let checksum = decode_to_vec(checksums.get(algo).unwrap()).unwrap(); - - let mut h = algo.new(); - let mut msg = Vec::with_capacity(h.size() * h.block_size()); - let mut sum = Vec::with_capacity(h.size()); + use crate::{ + bitrot::{new_bitrot_writer, BITROT_ALGORITHMS}, + disk::{endpoint::Endpoint, error::DiskError, new_disk, DiskOption}, + error::{Error, Result}, + store_api::BitrotAlgorithm, + }; - for i in (0..h.size()*h.block_size()).step_by(h.size()) { - h.update(&msg); - sum = h.finalize(); - msg.extend(sum.clone()); - h = algo.new(); + use super::{bitrot_writer_sum, new_bitrot_reader}; + + #[test] + fn bitrot_self_test() -> Result<()> { + let mut checksums = HashMap::new(); + checksums.insert( + BitrotAlgorithm::SHA256, + "a7677ff19e0182e4d52e3a3db727804abc82a5818749336369552e54b838b004", + ); + checksums.insert(BitrotAlgorithm::BLAKE2b512, "e519b7d84b1c3c917985f544773a35cf265dcab10948be3550320d156bab612124a5ae2ae5a8c73c0eea360f68b0e28136f26e858756dbfe7375a7389f26c669"); + checksums.insert( + BitrotAlgorithm::HighwayHash256, + "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b", + ); + checksums.insert( + BitrotAlgorithm::HighwayHash256S, + "c81c2386a1f565e805513d630d4e50ff26d11269b21c221cf50fc6c29d6ff75b", + ); + + let iter = [ + BitrotAlgorithm::SHA256, + BitrotAlgorithm::BLAKE2b512, + BitrotAlgorithm::HighwayHash256, + ]; + + for algo in iter.iter() { + if !algo.available() || *algo != BitrotAlgorithm::HighwayHash256 { + continue; + } + let checksum = decode_to_vec(checksums.get(algo).unwrap()).unwrap(); + + let mut h = algo.new(); + let mut msg = Vec::with_capacity(h.size() * h.block_size()); + let mut sum = Vec::with_capacity(h.size()); + + for _ in (0..h.size() * h.block_size()).step_by(h.size()) { + h.update(&msg); + sum = h.finalize(); + msg.extend(sum.clone()); + h = algo.new(); + } + + if checksum != sum { + println!("failed: {:?}, expect: {:?}, actual: {:?}", algo, checksum, sum); + return Err(Error::new(DiskError::FileCorrupt)); + } + println!("success: {:?}", algo); } - if checksum != sum { - println!("failed: {:?}, expect: {:?}, actual: {:?}", algo, checksum, sum); - return Err(Error::new(DiskError::FileCorrupt)); - } - println!("success: {:?}", algo); + Ok(()) } - Ok(()) + #[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?; + } + + Ok(()) + } + + async fn test_bitrot_reader_writer_algo(algo: BitrotAlgorithm) -> Result<()> { + let temp_dir = TempDir::new().unwrap().path().to_string_lossy().to_string(); + fs::create_dir_all(&temp_dir)?; + let volume = "testvol"; + let file_path = "testfile"; + + let ep = Endpoint::try_from(temp_dir.as_str())?; + let opt = DiskOption::default(); + let disk = new_disk(&ep, &opt).await?; + let _ = disk.make_volume(volume).await?; + let mut writer = new_bitrot_writer(disk.clone(), "", volume, file_path, 35, algo.clone(), 10); + + let _ = writer.write(b"aaaaaaaaaa").await?; + let _ = writer.write(b"aaaaaaaaaa").await?; + let _ = writer.write(b"aaaaaaaaaa").await?; + let _ = writer.write(b"aaaaa").await?; + + let mut reader = new_bitrot_reader(disk, b"", volume, file_path, 35, algo, &bitrot_writer_sum(&writer), 10); + let read_len = 10; + (_, _) = reader.read_at(0, read_len).await?; + (_, _) = reader.read_at(0, read_len).await?; + (_, _) = reader.read_at(0, read_len).await?; + (_, _) = reader.read_at(0, read_len / 2).await?; + + Ok(()) + } } diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 89a30746b..6408e8756 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -26,7 +26,7 @@ use protos::proto_gen::node_service::{ node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse, }; use serde::{Deserialize, Serialize}; -use std::{cmp::Ordering, collections::HashMap, fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc, usize}; +use std::{any::Any, cmp::Ordering, collections::HashMap, fmt::Debug, io::SeekFrom, path::PathBuf, sync::Arc, usize}; use time::OffsetDateTime; use tokio::{ fs::File, @@ -646,6 +646,10 @@ pub enum FileWriter { #[async_trait::async_trait] impl Write for FileWriter { + fn as_any(&self) -> &dyn Any { + self + } + async fn write(&mut self, buf: &[u8]) -> Result<()> { match self { Self::Local(local_file_writer) => local_file_writer.write(buf).await, @@ -666,6 +670,10 @@ impl LocalFileWriter { #[async_trait::async_trait] impl Write for LocalFileWriter { + fn as_any(&self) -> &dyn Any { + self + } + async fn write(&mut self, buf: &[u8]) -> Result<()> { let _ = self.inner.write(buf).await?; self.inner.flush().await?; @@ -714,6 +722,10 @@ impl RemoteFileWriter { #[async_trait::async_trait] impl Write for RemoteFileWriter { + fn as_any(&self) -> &dyn Any { + self + } + async fn write(&mut self, buf: &[u8]) -> Result<()> { let request = WriteRequest { disk: self.root.to_string_lossy().to_string(), diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index aae5ea128..4aac5c0ff 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -4,6 +4,7 @@ use bytes::Bytes; use futures::future::join_all; use futures::{Stream, StreamExt}; use reed_solomon_erasure::galois_8::ReedSolomon; +use std::any::Any; use std::fmt::Debug; use tokio::io::AsyncWriteExt; use tokio::io::DuplexStream; @@ -325,6 +326,7 @@ impl Erasure { #[async_trait::async_trait] pub trait Write { + fn as_any(&self) -> &dyn Any; async fn write(&mut self, buf: &[u8]) -> Result<()>; } diff --git a/ecstore/src/utils/os/linux.rs b/ecstore/src/utils/os/linux.rs index 254c71e6b..59b3951ca 100644 --- a/ecstore/src/utils/os/linux.rs +++ b/ecstore/src/utils/os/linux.rs @@ -1,79 +1,107 @@ -use nix::sys::{ - stat::{major, minor, stat}, - statfs::{statfs, FsType}, -}; +use nix::sys::stat::{self, stat}; +use nix::sys::statfs::{self, statfs, FsType}; +use std::io::{Error, ErrorKind}; +use std::path::Path; -use crate::{ - disk::Info, - error::{Error, Result}, -}; +use crate::{disk::Info, error::Result}; -use lazy_static::lazy_static; -use std::collections::HashMap; +/// returns total and free bytes available in a directory, e.g. `/`. +pub fn get_info(p: impl AsRef, _first_time: bool) -> std::io::Result { + let stat_fs = statfs(p.as_ref())?; -lazy_static! { - static ref FS_TYPE_TO_STRING_MAP: HashMap<&'static str, &'static str> = { - let mut m = HashMap::new(); - m.insert("1021994", "TMPFS"); - m.insert("137d", "EXT"); - m.insert("4244", "HFS"); - m.insert("4d44", "MSDOS"); - m.insert("52654973", "REISERFS"); - m.insert("5346544e", "NTFS"); - m.insert("58465342", "XFS"); - m.insert("61756673", "AUFS"); - m.insert("6969", "NFS"); - m.insert("ef51", "EXT2OLD"); - m.insert("ef53", "EXT4"); - m.insert("f15f", "ecryptfs"); - m.insert("794c7630", "overlayfs"); - m.insert("2fc12fc1", "zfs"); - m.insert("ff534d42", "cifs"); - m.insert("53464846", "wslfs"); - m + let bsize = stat_fs.block_size() as u64; + let bfree = stat_fs.blocks_free() as u64; + let bavail = stat_fs.blocks_available() as u64; + let blocks = stat_fs.blocks() as u64; + + let reserved = match bfree.checked_sub(bavail) { + Some(reserved) => reserved, + None => { + return Err(Error::new( + ErrorKind::Other, + format!( + "detected f_bavail space ({}) > f_bfree space ({}), fs corruption at ({}). please run 'fsck'", + bavail, + bfree, + p.as_ref().display() + ), + )) + } }; -} -fn get_fs_type(ftype: FsType) -> String { - let binding = format!("{:?}", ftype); - let fs_type_hex = binding.as_str(); - match FS_TYPE_TO_STRING_MAP.get(fs_type_hex) { - Some(fs_type_string) => fs_type_string.to_string(), - None => "UNKNOWN".to_string(), - } -} + let total = match blocks.checked_sub(reserved) { + Some(total) => total * bsize, + None => { + return Err(Error::new( + ErrorKind::Other, + format!( + "detected reserved space ({}) > blocks space ({}), fs corruption at ({}). please run 'fsck'", + reserved, + blocks, + p.as_ref().display() + ), + )) + } + }; + + let free = bavail * bsize; + let used = match total.checked_sub(free) { + Some(used) => used, + None => { + return Err(Error::new( + ErrorKind::Other, + format!( + "detected free space ({}) > total drive space ({}), fs corruption at ({}). please run 'fsck'", + free, + total, + p.as_ref().display() + ), + )) + } + }; + + let st = stat(p.as_ref())?; + + Ok(Info { + total, + free, + used, + files: stat_fs.files(), + ffree: stat_fs.files_free(), + fstype: get_fs_type(stat_fs.filesystem_type()).to_string(), + + major: stat::major(st.st_dev), + minor: stat::minor(st.st_dev), -pub fn get_info(path: &str, first_time: bool) -> Result { - let statfs = statfs(path)?; - let reserved_blocks = statfs.blocks_free() - statfs.blocks_available(); - let mut info = Info { - total: statfs.block_size() as u64 * (statfs.blocks() - reserved_blocks), - free: statfs.blocks() as u64 * statfs.blocks_available(), - files: statfs.files(), - ffree: statfs.files_free(), - fstype: get_fs_type(statfs.filesystem_type()), ..Default::default() - }; + }) +} - let stat = stat(path)?; - let dev_id = stat.st_dev as u64; - info.major = major(dev_id); - info.minor = minor(dev_id); +/// returns the filesystem type of the underlying mounted filesystem +/// +/// TODO The following mapping could not find the corresponding constant in `nix`: +/// +/// "137d" => "EXT", +/// "4244" => "HFS", +/// "5346544e" => "NTFS", +/// "61756673" => "AUFS", +/// "ef51" => "EXT2OLD", +/// "2fc12fc1" => "zfs", +/// "ff534d42" => "cifs", +/// "53464846" => "wslfs", +fn get_fs_type(fs_type: FsType) -> &'static str { + match fs_type { + statfs::TMPFS_MAGIC => "TMPFS", + statfs::MSDOS_SUPER_MAGIC => "MSDOS", + statfs::XFS_SUPER_MAGIC => "XFS", + statfs::NFS_SUPER_MAGIC => "NFS", + statfs::EXT4_SUPER_MAGIC => "EXT4", + statfs::ECRYPTFS_SUPER_MAGIC => "ecryptfs", + statfs::OVERLAYFS_SUPER_MAGIC => "overlayfs", + statfs::REISERFS_SUPER_MAGIC => "REISERFS", - if info.free > info.total { - return Err(Error::from_string(format!( - "detected free space {} > total drive space {}, fs corruption at {}. please run 'fsck'", - info.free, info.total, path - ))); + _ => "UNKNOWN", } - - info.used = info.total - info.free; - - if first_time { - // todo - } - - Ok(info) } pub fn same_disk(disk1: &str, disk2: &str) -> Result { @@ -81,4 +109,4 @@ pub fn same_disk(disk1: &str, disk2: &str) -> Result { let stat2 = stat(disk2)?; Ok(stat1.st_dev == stat2.st_dev) -} +} \ No newline at end of file