pub mod endpoint; pub mod error; pub mod format; pub mod local; pub mod os; pub mod remote; pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys"; pub const RUSTFS_META_MULTIPART_BUCKET: &str = ".rustfs.sys/multipart"; pub const RUSTFS_META_TMP_BUCKET: &str = ".rustfs.sys/tmp"; pub const RUSTFS_META_TMP_DELETED_BUCKET: &str = ".rustfs.sys/tmp/.trash"; pub const BUCKET_META_PREFIX: &str = "buckets"; pub const FORMAT_CONFIG_FILE: &str = "format.json"; pub const STORAGE_FORMAT_FILE: &str = "xl.meta"; pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp"; use crate::{ bucket::{metadata_sys::get_versioning_config, versioning::VersioningApi}, file_meta::{merge_file_meta_versions, FileMeta, FileMetaShallowVersion, VersionType}, heal::{ data_scanner::ShouldSleepFn, data_usage_cache::{DataUsageCache, DataUsageEntry}, heal_commands::{HealScanMode, HealingTracker}, }, io::{FileReader, FileWriter}, store_api::{FileInfo, ObjectInfo, RawFileInfo}, utils::path::SLASH_SEPARATOR, }; use common::error::{Error, Result}; use endpoint::Endpoint; use error::DiskError; use local::LocalDisk; use madmin::info_commands::DiskMetrics; use remote::RemoteDisk; use serde::{Deserialize, Serialize}; use std::{cmp::Ordering, fmt::Debug, path::PathBuf, sync::Arc}; use time::OffsetDateTime; use tokio::{io::AsyncWrite, sync::mpsc::Sender}; use tracing::warn; use uuid::Uuid; pub type DiskStore = Arc; #[derive(Debug)] pub enum Disk { Local(Box), Remote(Box), } #[async_trait::async_trait] impl DiskAPI for Disk { #[tracing::instrument(skip(self))] fn to_string(&self) -> String { match self { Disk::Local(local_disk) => local_disk.to_string(), Disk::Remote(remote_disk) => remote_disk.to_string(), } } #[tracing::instrument(skip(self))] async fn is_online(&self) -> bool { match self { Disk::Local(local_disk) => local_disk.is_online().await, Disk::Remote(remote_disk) => remote_disk.is_online().await, } } #[tracing::instrument(skip(self))] fn is_local(&self) -> bool { match self { Disk::Local(local_disk) => local_disk.is_local(), Disk::Remote(remote_disk) => remote_disk.is_local(), } } #[tracing::instrument(skip(self))] fn host_name(&self) -> String { match self { Disk::Local(local_disk) => local_disk.host_name(), Disk::Remote(remote_disk) => remote_disk.host_name(), } } #[tracing::instrument(skip(self))] fn endpoint(&self) -> Endpoint { match self { Disk::Local(local_disk) => local_disk.endpoint(), Disk::Remote(remote_disk) => remote_disk.endpoint(), } } #[tracing::instrument(skip(self))] async fn close(&self) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.close().await, Disk::Remote(remote_disk) => remote_disk.close().await, } } #[tracing::instrument(skip(self))] async fn get_disk_id(&self) -> Result> { match self { Disk::Local(local_disk) => local_disk.get_disk_id().await, Disk::Remote(remote_disk) => remote_disk.get_disk_id().await, } } #[tracing::instrument(skip(self))] async fn set_disk_id(&self, id: Option) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.set_disk_id(id).await, Disk::Remote(remote_disk) => remote_disk.set_disk_id(id).await, } } #[tracing::instrument(skip(self))] fn path(&self) -> PathBuf { match self { Disk::Local(local_disk) => local_disk.path(), Disk::Remote(remote_disk) => remote_disk.path(), } } #[tracing::instrument(skip(self))] fn get_disk_location(&self) -> DiskLocation { match self { Disk::Local(local_disk) => local_disk.get_disk_location(), Disk::Remote(remote_disk) => remote_disk.get_disk_location(), } } #[tracing::instrument(skip(self))] async fn make_volume(&self, volume: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.make_volume(volume).await, Disk::Remote(remote_disk) => remote_disk.make_volume(volume).await, } } #[tracing::instrument(skip(self))] async fn make_volumes(&self, volumes: Vec<&str>) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.make_volumes(volumes).await, Disk::Remote(remote_disk) => remote_disk.make_volumes(volumes).await, } } #[tracing::instrument(skip(self))] async fn list_volumes(&self) -> Result> { match self { Disk::Local(local_disk) => local_disk.list_volumes().await, Disk::Remote(remote_disk) => remote_disk.list_volumes().await, } } #[tracing::instrument(skip(self))] async fn stat_volume(&self, volume: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.stat_volume(volume).await, Disk::Remote(remote_disk) => remote_disk.stat_volume(volume).await, } } #[tracing::instrument(skip(self))] async fn delete_volume(&self, volume: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_volume(volume).await, Disk::Remote(remote_disk) => remote_disk.delete_volume(volume).await, } } #[tracing::instrument(skip(self, wr))] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.walk_dir(opts, wr).await, Disk::Remote(remote_disk) => remote_disk.walk_dir(opts, wr).await, } } #[tracing::instrument(skip(self))] async fn delete_version( &self, volume: &str, path: &str, fi: FileInfo, force_del_marker: bool, opts: DeleteOptions, ) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_version(volume, path, fi, force_del_marker, opts).await, Disk::Remote(remote_disk) => remote_disk.delete_version(volume, path, fi, force_del_marker, opts).await, } } #[tracing::instrument(skip(self))] async fn delete_versions( &self, volume: &str, versions: Vec, opts: DeleteOptions, ) -> Result>> { match self { Disk::Local(local_disk) => local_disk.delete_versions(volume, versions, opts).await, Disk::Remote(remote_disk) => remote_disk.delete_versions(volume, versions, opts).await, } } #[tracing::instrument(skip(self))] async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete_paths(volume, paths).await, Disk::Remote(remote_disk) => remote_disk.delete_paths(volume, paths).await, } } #[tracing::instrument(skip(self))] async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_metadata(_org_volume, volume, path, fi).await, Disk::Remote(remote_disk) => remote_disk.write_metadata(_org_volume, volume, path, fi).await, } } #[tracing::instrument(skip(self))] async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.update_metadata(volume, path, fi, opts).await, Disk::Remote(remote_disk) => remote_disk.update_metadata(volume, path, fi, opts).await, } } #[tracing::instrument(level = "debug", skip(self))] async fn read_version( &self, _org_volume: &str, volume: &str, path: &str, version_id: &str, opts: &ReadOptions, ) -> Result { match self { Disk::Local(local_disk) => local_disk.read_version(_org_volume, volume, path, version_id, opts).await, Disk::Remote(remote_disk) => remote_disk.read_version(_org_volume, volume, path, version_id, opts).await, } } #[tracing::instrument(skip(self))] async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result { match self { Disk::Local(local_disk) => local_disk.read_xl(volume, path, read_data).await, Disk::Remote(remote_disk) => remote_disk.read_xl(volume, path, read_data).await, } } #[tracing::instrument(skip(self, fi))] async fn rename_data( &self, src_volume: &str, src_path: &str, fi: FileInfo, dst_volume: &str, dst_path: &str, ) -> Result { match self { Disk::Local(local_disk) => local_disk.rename_data(src_volume, src_path, fi, dst_volume, dst_path).await, Disk::Remote(remote_disk) => remote_disk.rename_data(src_volume, src_path, fi, dst_volume, dst_path).await, } } #[tracing::instrument(skip(self))] async fn list_dir(&self, _origvolume: &str, volume: &str, _dir_path: &str, _count: i32) -> Result> { match self { Disk::Local(local_disk) => local_disk.list_dir(_origvolume, volume, _dir_path, _count).await, Disk::Remote(remote_disk) => remote_disk.list_dir(_origvolume, volume, _dir_path, _count).await, } } #[tracing::instrument(skip(self))] async fn read_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file(volume, path).await, Disk::Remote(remote_disk) => remote_disk.read_file(volume, path).await, } } #[tracing::instrument(skip(self))] async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.read_file_stream(volume, path, offset, length).await, Disk::Remote(remote_disk) => remote_disk.read_file_stream(volume, path, offset, length).await, } } #[tracing::instrument(skip(self))] async fn append_file(&self, volume: &str, path: &str) -> Result { match self { Disk::Local(local_disk) => local_disk.append_file(volume, path).await, Disk::Remote(remote_disk) => remote_disk.append_file(volume, path).await, } } #[tracing::instrument(skip(self))] async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result { match self { Disk::Local(local_disk) => local_disk.create_file(_origvolume, volume, path, _file_size).await, Disk::Remote(remote_disk) => remote_disk.create_file(_origvolume, volume, path, _file_size).await, } } #[tracing::instrument(skip(self))] async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await, Disk::Remote(remote_disk) => remote_disk.rename_file(src_volume, src_path, dst_volume, dst_path).await, } } #[tracing::instrument(skip(self))] async fn rename_part(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str, meta: Vec) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.rename_part(src_volume, src_path, dst_volume, dst_path, meta).await, Disk::Remote(remote_disk) => { remote_disk .rename_part(src_volume, src_path, dst_volume, dst_path, meta) .await } } } #[tracing::instrument(skip(self))] async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.delete(volume, path, opt).await, Disk::Remote(remote_disk) => remote_disk.delete(volume, path, opt).await, } } #[tracing::instrument(skip(self))] async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { match self { Disk::Local(local_disk) => local_disk.verify_file(volume, path, fi).await, Disk::Remote(remote_disk) => remote_disk.verify_file(volume, path, fi).await, } } #[tracing::instrument(skip(self))] async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result { match self { Disk::Local(local_disk) => local_disk.check_parts(volume, path, fi).await, Disk::Remote(remote_disk) => remote_disk.check_parts(volume, path, fi).await, } } #[tracing::instrument(skip(self))] async fn read_multiple(&self, req: ReadMultipleReq) -> Result> { match self { Disk::Local(local_disk) => local_disk.read_multiple(req).await, Disk::Remote(remote_disk) => remote_disk.read_multiple(req).await, } } #[tracing::instrument(skip(self))] async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()> { match self { Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await, Disk::Remote(remote_disk) => remote_disk.write_all(volume, path, data).await, } } #[tracing::instrument(skip(self))] async fn read_all(&self, volume: &str, path: &str) -> Result> { match self { Disk::Local(local_disk) => local_disk.read_all(volume, path).await, Disk::Remote(remote_disk) => remote_disk.read_all(volume, path).await, } } #[tracing::instrument(skip(self))] async fn disk_info(&self, opts: &DiskInfoOptions) -> Result { match self { Disk::Local(local_disk) => local_disk.disk_info(opts).await, Disk::Remote(remote_disk) => remote_disk.disk_info(opts).await, } } #[tracing::instrument(skip(self, cache, we_sleep, scan_mode))] async fn ns_scanner( &self, cache: &DataUsageCache, updates: Sender, scan_mode: HealScanMode, we_sleep: ShouldSleepFn, ) -> Result { match self { Disk::Local(local_disk) => local_disk.ns_scanner(cache, updates, scan_mode, we_sleep).await, Disk::Remote(remote_disk) => remote_disk.ns_scanner(cache, updates, scan_mode, we_sleep).await, } } #[tracing::instrument(skip(self))] async fn healing(&self) -> Option { match self { Disk::Local(local_disk) => local_disk.healing().await, Disk::Remote(remote_disk) => remote_disk.healing().await, } } } pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result { if ep.is_local { let s = LocalDisk::new(ep, opt.cleanup).await?; Ok(Arc::new(Disk::Local(Box::new(s)))) } else { let remote_disk = RemoteDisk::new(ep, opt).await?; Ok(Arc::new(Disk::Remote(Box::new(remote_disk)))) } } #[async_trait::async_trait] pub trait DiskAPI: Debug + Send + Sync + 'static { fn to_string(&self) -> String; async fn is_online(&self) -> bool; fn is_local(&self) -> bool; // LastConn fn host_name(&self) -> String; fn endpoint(&self) -> Endpoint; async fn close(&self) -> Result<()>; async fn get_disk_id(&self) -> Result>; async fn set_disk_id(&self, id: Option) -> Result<()>; fn path(&self) -> PathBuf; fn get_disk_location(&self) -> DiskLocation; // Healing // DiskInfo // NSScanner // Volume operations. async fn make_volume(&self, volume: &str) -> Result<()>; async fn make_volumes(&self, volume: Vec<&str>) -> Result<()>; async fn list_volumes(&self) -> Result>; async fn stat_volume(&self, volume: &str) -> Result; async fn delete_volume(&self, volume: &str) -> Result<()>; // 并发边读边写 w <- MetaCacheEntry async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()>; // Metadata operations async fn delete_version( &self, volume: &str, path: &str, fi: FileInfo, force_del_marker: bool, opts: DeleteOptions, ) -> Result<()>; async fn delete_versions( &self, volume: &str, versions: Vec, opts: DeleteOptions, ) -> Result>>; async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()>; async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>; async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()>; async fn read_version( &self, org_volume: &str, volume: &str, path: &str, version_id: &str, opts: &ReadOptions, ) -> Result; async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result; async fn rename_data( &self, src_volume: &str, src_path: &str, file_info: FileInfo, dst_volume: &str, dst_path: &str, ) -> Result; // File operations. // 读目录下的所有文件、目录 async fn list_dir(&self, origvolume: &str, volume: &str, dir_path: &str, count: i32) -> Result>; async fn read_file(&self, volume: &str, path: &str) -> Result; async fn read_file_stream(&self, volume: &str, path: &str, offset: usize, length: usize) -> Result; async fn append_file(&self, volume: &str, path: &str) -> Result; async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result; // 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) -> Result<()>; async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()>; // VerifyFile async fn verify_file(&self, volume: &str, path: &str, fi: &FileInfo) -> Result; // CheckParts async fn check_parts(&self, volume: &str, path: &str, fi: &FileInfo) -> Result; // StatInfoFile // ReadParts async fn read_multiple(&self, req: ReadMultipleReq) -> Result>; // CleanAbandonedData async fn write_all(&self, volume: &str, path: &str, data: Vec) -> Result<()>; #[must_use] async fn read_all(&self, volume: &str, path: &str) -> Result>; async fn disk_info(&self, opts: &DiskInfoOptions) -> Result; async fn ns_scanner( &self, cache: &DataUsageCache, updates: Sender, scan_mode: HealScanMode, we_sleep: ShouldSleepFn, ) -> Result; async fn healing(&self) -> Option; } #[derive(Debug, Default, Serialize, Deserialize)] pub struct CheckPartsResp { pub results: Vec, } #[derive(Debug, Serialize, Deserialize, Default)] pub struct UpdateMetadataOpts { pub no_persistence: bool, } pub struct DiskLocation { pub pool_idx: Option, pub set_idx: Option, pub disk_idx: Option, } impl DiskLocation { pub fn valid(&self) -> bool { self.pool_idx.is_some() && self.set_idx.is_some() && self.disk_idx.is_some() } } #[derive(Debug, Default, Serialize, Deserialize)] pub struct DiskInfoOptions { pub disk_id: String, pub metrics: bool, pub noop: bool, } #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq, Eq)] pub struct DiskInfo { pub total: u64, pub free: u64, pub used: u64, pub used_inodes: u64, pub free_inodes: u64, pub major: u64, pub minor: u64, pub nr_requests: u64, pub fs_type: String, pub root_disk: bool, pub healing: bool, pub scanning: bool, pub endpoint: String, pub mount_path: String, pub id: String, pub rotational: bool, pub metrics: DiskMetrics, pub error: String, } #[derive(Clone, Debug, Default)] pub struct Info { pub total: u64, pub free: u64, pub used: u64, pub files: u64, pub ffree: u64, pub fstype: String, pub major: u64, pub minor: u64, pub name: String, pub rotational: bool, pub nrrequests: u64, } #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct FileInfoVersions { // Name of the volume. pub volume: String, // Name of the file. pub name: String, // Represents the latest mod time of the // latest version. pub latest_mod_time: Option, pub versions: Vec, pub free_versions: Vec, } impl FileInfoVersions { pub fn find_version_index(&self, v: &str) -> Option { if v.is_empty() { return None; } let vid = Uuid::parse_str(v).unwrap_or_default(); self.versions.iter().position(|v| v.version_id == Some(vid)) } } #[derive(Debug, Default, Clone, Serialize, Deserialize)] pub struct WalkDirOptions { // Bucket to scanner pub bucket: String, // Directory inside the bucket. pub base_dir: String, // Do a full recursive scan. pub recursive: bool, // ReportNotFound will return errFileNotFound if all disks reports the BaseDir cannot be found. pub report_notfound: bool, // FilterPrefix will only return results with given prefix within folder. // Should never contain a slash. pub filter_prefix: Option, // ForwardTo will forward to the given object path. pub forward_to: Option, // Limit the number of returned objects if > 0. pub limit: i32, // DiskID contains the disk ID of the disk. // Leave empty to not check disk ID. pub disk_id: String, } #[derive(Clone, Debug, Default)] pub struct MetadataResolutionParams { pub dir_quorum: usize, pub obj_quorum: usize, pub requested_versions: usize, pub bucket: String, pub strict: bool, pub candidates: Vec>, } #[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)] pub struct MetaCacheEntry { // name is the full name of the object including prefixes pub name: String, // Metadata. If none is present it is not an object but only a prefix. // Entries without metadata will only be present in non-recursive scans. pub metadata: Vec, // cached contains the metadata if decoded. pub cached: Option, // Indicates the entry can be reused and only one reference to metadata is expected. pub reusable: bool, } impl MetaCacheEntry { pub fn marshal_msg(&self) -> Result> { let mut wr = Vec::new(); rmp::encode::write_bool(&mut wr, true)?; rmp::encode::write_str(&mut wr, &self.name)?; rmp::encode::write_bin(&mut wr, &self.metadata)?; 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_delete_marker(&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() { return FileMeta::is_latest_delete_marker(meta); } } 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 { if self.is_dir() { 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(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.to_fileinfo(bucket, self.name.as_str(), "", false, false)?; return Ok(fi); } let mut fm = FileMeta::new(); fm.unmarshal_msg(&self.metadata)?; let fi = fm.to_fileinfo(bucket, self.name.as_str(), "", false, false)?; Ok(fi) } pub fn file_info_versions(&self, bucket: &str) -> Result { if self.is_dir() { return Ok(FileInfoVersions { volume: bucket.to_string(), name: self.name.clone(), versions: vec![FileInfo { volume: bucket.to_string(), name: self.name.clone(), ..Default::default() }], ..Default::default() }); } let mut fm = FileMeta::new(); fm.unmarshal_msg(&self.metadata)?; fm.into_file_info_versions(bucket, self.name.as_str(), false) } pub fn matches(&self, other: Option<&MetaCacheEntry>, strict: bool) -> (Option, bool) { if other.is_none() { return (None, false); } let other = other.unwrap(); let mut prefer = None; if self.name != other.name { if self.name < other.name { return (Some(self.clone()), false); } return (Some(other.clone()), false); } if other.is_dir() || self.is_dir() { if self.is_dir() { return (Some(self.clone()), other.is_dir() == self.is_dir()); } return (Some(other.clone()), other.is_dir() == self.is_dir()); } let self_vers = match &self.cached { Some(file_meta) => file_meta.clone(), None => match FileMeta::load(&self.metadata) { Ok(meta) => meta, Err(_) => { return (None, false); } }, }; let other_vers = match &other.cached { Some(file_meta) => file_meta.clone(), None => match FileMeta::load(&other.metadata) { Ok(meta) => meta, Err(_) => { return (None, false); } }, }; if self_vers.versions.len() != other_vers.versions.len() { match self_vers.lastest_mod_time().cmp(&other_vers.lastest_mod_time()) { Ordering::Greater => { return (Some(self.clone()), false); } Ordering::Less => { return (Some(other.clone()), false); } _ => {} } if self_vers.versions.len() > other_vers.versions.len() { return (Some(self.clone()), false); } return (Some(other.clone()), false); } for (s_version, o_version) in self_vers.versions.iter().zip(other_vers.versions.iter()) { if s_version.header != o_version.header { if s_version.header.has_ec() != o_version.header.has_ec() { // One version has EC and the other doesn't - may have been written later. // Compare without considering EC. let (mut a, mut b) = (s_version.header.clone(), o_version.header.clone()); (a.ec_n, a.ec_m, b.ec_n, b.ec_m) = (0, 0, 0, 0); if a == b { continue; } } if !strict && s_version.header.matches_not_strict(&o_version.header) { if prefer.is_none() { if s_version.header.sorts_before(&o_version.header) { prefer = Some(self.clone()); } else { prefer = Some(other.clone()); } } continue; } if prefer.is_some() { return (prefer, false); } if s_version.header.sorts_before(&o_version.header) { return (Some(self.clone()), false); } return (Some(other.clone()), false); } } if prefer.is_none() { prefer = Some(self.clone()); } (prefer, true) } pub fn xl_meta(&mut self) -> Result { 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) } } } #[derive(Debug, Default)] pub struct MetaCacheEntries(pub Vec>); impl MetaCacheEntries { #[allow(clippy::should_implement_trait)] pub fn as_ref(&self) -> &[Option] { &self.0 } pub fn resolve(&self, mut params: MetadataResolutionParams) -> Option { if self.0.is_empty() { warn!("decommission_pool: entries resolve empty"); return None; } let mut dir_exists = 0; let mut selected = None; params.candidates.clear(); let mut objs_agree = 0; let mut objs_valid = 0; for entry in self.0.iter().flatten() { let mut entry = entry.clone(); warn!("decommission_pool: entries resolve entry {:?}", entry.name); if entry.name.is_empty() { continue; } if entry.is_dir() { dir_exists += 1; selected = Some(entry.clone()); warn!("decommission_pool: entries resolve entry dir {:?}", entry.name); continue; } let xl = match entry.xl_meta() { Ok(xl) => xl, Err(e) => { warn!("decommission_pool: entries resolve entry xl_meta {:?}", e); continue; } }; objs_valid += 1; params.candidates.push(xl.versions.clone()); if selected.is_none() { selected = Some(entry.clone()); objs_agree = 1; warn!("decommission_pool: entries resolve entry selected {:?}", entry.name); continue; } if let (prefer, true) = entry.matches(selected.as_ref(), params.strict) { selected = prefer; objs_agree += 1; warn!("decommission_pool: entries resolve entry prefer {:?}", entry.name); continue; } } let Some(selected) = selected else { warn!("decommission_pool: entries resolve entry no selected"); return None; }; if selected.is_dir() && dir_exists >= params.dir_quorum { warn!("decommission_pool: entries resolve entry dir selected {:?}", selected.name); return Some(selected); } // If we would never be able to reach read quorum. if objs_valid < params.obj_quorum { warn!( "decommission_pool: entries resolve entry not enough objects {} < {}", objs_valid, params.obj_quorum ); return None; } if objs_agree == objs_valid { warn!("decommission_pool: entries resolve entry all agree {} == {}", objs_agree, objs_valid); return Some(selected); } let Some(cached) = selected.cached else { warn!("decommission_pool: entries resolve entry no cached"); return None; }; let versions = merge_file_meta_versions(params.obj_quorum, params.strict, params.requested_versions, ¶ms.candidates); if versions.is_empty() { warn!("decommission_pool: entries resolve entry no versions"); return None; } let metadata = match cached.marshal_msg() { Ok(meta) => meta, Err(e) => { warn!("decommission_pool: entries resolve entry marshal_msg {:?}", e); return None; } }; // Merge if we have disagreement. // Create a new merged result. let new_selected = MetaCacheEntry { name: selected.name.clone(), cached: Some(FileMeta { meta_ver: cached.meta_ver, versions, ..Default::default() }), reusable: true, metadata, }; warn!("decommission_pool: entries resolve entry selected {:?}", new_selected.name); Some(new_selected) } pub fn first_found(&self) -> (Option, usize) { (self.0.iter().find(|x| x.is_some()).cloned().unwrap_or_default(), self.0.len()) } } #[derive(Debug, Default)] pub struct MetaCacheEntriesSortedResult { pub entries: Option, pub err: Option, } // 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, pub reuse: bool, pub last_skipped_entry: Option, } 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) { 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) -> Vec { 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, after_v: Option, ) -> Vec { 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 } } #[derive(Clone, Debug, Default)] pub struct DiskOption { pub cleanup: bool, pub health_check: bool, } #[derive(Debug, Default, Serialize, Deserialize)] pub struct RenameDataResp { pub old_data_dir: Option, pub sign: Option>, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct DeleteOptions { pub recursive: bool, pub immediate: bool, pub undo_write: bool, pub old_data_dir: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] pub struct ReadMultipleReq { pub bucket: String, pub prefix: String, pub files: Vec, pub max_size: usize, pub metadata_only: bool, pub abort404: bool, pub max_results: usize, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct ReadMultipleResp { pub bucket: String, pub prefix: String, pub file: String, pub exists: bool, pub error: String, pub data: Vec, pub mod_time: Option, } #[derive(Debug, Deserialize, Serialize)] pub struct VolumeInfo { pub name: String, pub created: Option, } #[derive(Deserialize, Serialize, Debug, Default)] pub struct ReadOptions { pub incl_free_versions: bool, pub read_data: bool, pub healing: bool, }