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

change objectinfo metadata use HashMap

Browse Source 
add copyobject compresss support

fix listobject size use actual_size
This commit is contained in:
weisd
2025-06-23 11:37:42 +08:00
parent dc93398768
commit 34ec13e532
9 changed files with 143 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/cmd/bucket_replication.rs
+20 -33
View File
@@ -417,21 +417,16 @@ pub async fn get_heal_replicate_object_info(
// .map(|(k, v)| (k.clone(), v.clone()))
// .collect::<HashMap<String, String>>()
// .collect();
let to_replace: Vec<(String, String)> = match &user_defined {
Some(map) => map
.iter()
.filter(|(k, _)| k.eq_ignore_ascii_case(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_RESET)))
.map(|(k, v)| (k.clone(), v.clone()))
.collect(),
None => Vec::new(),
};
let to_replace: Vec<(String, String)> = user_defined
.iter()
.filter(|(k, _)| k.eq_ignore_ascii_case(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_RESET)))
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
// 第二步：apply 修改
for (k, v) in to_replace {
if let Some(mp) = user_defined.as_mut() {
mp.remove(&k);
mp.insert(target_reset_header(&rcfg.role), v);
}
user_defined.remove(&k);
user_defined.insert(target_reset_header(&rcfg.role), v);
}
}
//}
@@ -468,13 +463,8 @@ pub async fn get_heal_replicate_object_info(
mod_time: oi.mod_time,
..Default::default()
};
let repoptions = get_must_replicate_options(
mt2.as_ref().unwrap_or(&HashMap::new()),
"",
ReplicationStatusType::Unknown,
ReplicationType::ObjectReplicationType,
&opts,
);
let repoptions =
get_must_replicate_options(&mt2, "", ReplicationStatusType::Unknown, ReplicationType::ObjectReplicationType, &opts);
let decision = must_replicate(&oi.bucket, &oi.name, &repoptions).await;
error!("decision:");
@@ -490,11 +480,10 @@ pub async fn get_heal_replicate_object_info(
// .and_then(|v| DateTime::parse_from_rfc3339(v).ok())
// .map(|dt| dt.with_timezone(&Utc));
let tm = user_defined.as_ref().and_then(|map| {
map.get(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_TIMESTAMP))
.and_then(|v| DateTime::parse_from_rfc3339(v).ok())
.map(|dt| dt.with_timezone(&Utc))
});
let tm = user_defined
.get(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_TIMESTAMP))
.and_then(|v| DateTime::parse_from_rfc3339(v).ok())
.map(|dt| dt.with_timezone(&Utc));
let mut rstate = oi.replication_state();
rstate.replicate_decision_str = dsc.to_string();
@@ -503,8 +492,6 @@ pub async fn get_heal_replicate_object_info(
let key = format!("{}{}", RESERVED_METADATA_PREFIX_LOWER, REPLICATION_TIMESTAMP);
let tm: Option<DateTime<Utc>> = user_defined
.as_ref()
.unwrap()
.get(&key)
.and_then(|v| DateTime::parse_from_rfc3339(v).ok())
.map(|dt| dt.with_timezone(&Utc));
@@ -1731,14 +1718,14 @@ impl TraitForObjectInfo for ObjectInfo {
rs.purge_targets = version_purge_statuses_map("");
// Process reset statuses map
if self.user_defined.is_some() {
for (k, v) in self.user_defined.as_ref().unwrap() {
if k.starts_with(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_RESET)) {
let arn = k.trim_start_matches(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_RESET));
rs.reset_statuses_map.insert(arn.to_string(), v.clone());
}
for (k, v) in self.user_defined.iter() {
if k.starts_with(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_RESET)) {
let arn = k.trim_start_matches(&(RESERVED_METADATA_PREFIX_LOWER.to_owned() + REPLICATION_RESET));
rs.reset_statuses_map.insert(arn.to_string(), v.clone());
}
}
rs
}
}
@@ -2021,7 +2008,7 @@ impl ReplicateObjectInfo {
size: self.size,
actual_size: self.actual_size,
is_dir: false,
user_defined: None, // 可以按需从别处导入
user_defined: HashMap::new(), // 可以按需从别处导入
parity_blocks: 0,
data_blocks: 0,
version_id: Uuid::try_parse(&self.version_id).ok(),
ecstore/src/heal/data_scanner.rs
+1 -2
View File
@@ -573,8 +573,7 @@ impl ScannerItem {
let mut oi = oi.clone();
oi.replication_status = ReplicationStatusType::from(
oi.user_defined
.as_ref()
.and_then(|map| map.get("x-amz-bucket-replication-status"))
.get("x-amz-bucket-replication-status")
.unwrap_or(&"PENDING".to_string()),
);
info!("apply status is: {:?}", oi.replication_status);
ecstore/src/set_disk.rs
+9 -13
View File
@@ -3816,7 +3816,7 @@ impl ObjectIO for SetDisks {
// _ns = Some(ns_lock);
// }
let mut user_defined = opts.user_defined.clone().unwrap_or_default();
let mut user_defined = opts.user_defined.clone();
let sc_parity_drives = {
if let Some(sc) = GLOBAL_StorageClass.get() {
@@ -4152,19 +4152,17 @@ impl StorageAPI for SetDisks {
};
let inline_data = fi.inline_data();
fi.metadata = src_info.user_defined.as_ref().cloned().unwrap_or_default();
fi.metadata = src_info.user_defined.clone();
if let Some(ud) = src_info.user_defined.as_mut() {
if let Some(etag) = &src_info.etag {
ud.insert("etag".to_owned(), etag.clone());
}
if let Some(etag) = &src_info.etag {
fi.metadata.insert("etag".to_owned(), etag.clone());
}
let mod_time = OffsetDateTime::now_utc();
for fi in metas.iter_mut() {
if fi.is_valid() {
fi.metadata = src_info.user_defined.as_ref().cloned().unwrap_or_default();
fi.metadata = src_info.user_defined.clone();
fi.mod_time = Some(mod_time);
fi.version_id = version_id;
fi.versioned = src_opts.versioned || src_opts.version_suspended;
@@ -4439,10 +4437,8 @@ impl StorageAPI for SetDisks {
let obj_info = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended);
if let Some(ud) = obj_info.user_defined.as_ref() {
for (k, v) in ud {
fi.metadata.insert(k.clone(), v.clone());
}
for (k, v) in obj_info.user_defined {
fi.metadata.insert(k, v);
}
if let Some(mt) = &opts.eval_metadata {
@@ -4851,7 +4847,7 @@ impl StorageAPI for SetDisks {
let disks = disks.clone();
let mut user_defined = opts.user_defined.clone().unwrap_or_default();
let mut user_defined = opts.user_defined.clone();
if let Some(ref etag) = opts.preserve_etag {
user_defined.insert("etag".to_owned(), etag.clone());
@@ -5111,7 +5107,7 @@ impl StorageAPI for SetDisks {
// etag
let etag = {
if let Some(Some(etag)) = opts.user_defined.as_ref().map(|v| v.get("etag")) {
if let Some(etag) = opts.user_defined.get("etag") {
etag.clone()
} else {
get_complete_multipart_md5(&uploaded_parts)
ecstore/src/store_api.rs
+15 -18
View File
@@ -283,7 +283,7 @@ pub struct ObjectOptions {
pub data_movement: bool,
pub src_pool_idx: usize,
pub user_defined: Option<HashMap<String, String>>,
pub user_defined: HashMap<String, String>,
pub preserve_etag: Option<String>,
pub metadata_chg: bool,
@@ -357,7 +357,7 @@ pub struct ObjectInfo {
// Actual size is the real size of the object uploaded by client.
pub actual_size: i64,
pub is_dir: bool,
pub user_defined: Option<HashMap<String, String>>,
pub user_defined: HashMap<String, String>,
pub parity_blocks: usize,
pub data_blocks: usize,
pub version_id: Option<Uuid>,
@@ -418,18 +418,15 @@ impl Clone for ObjectInfo {
impl ObjectInfo {
pub fn is_compressed(&self) -> bool {
if let Some(meta) = &self.user_defined {
meta.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER))
} else {
false
}
self.user_defined
.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER))
}
pub fn is_compressed_ok(&self) -> Result<(CompressionAlgorithm, bool)> {
let scheme = self
.user_defined
.as_ref()
.and_then(|meta| meta.get(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER)).cloned());
.get(&format!("{}compression", RESERVED_METADATA_PREFIX_LOWER))
.cloned();
if let Some(scheme) = scheme {
let algorithm = CompressionAlgorithm::from_str(&scheme)?;
@@ -449,16 +446,16 @@ impl ObjectInfo {
}
if self.is_compressed() {
if let Some(meta) = &self.user_defined {
if let Some(size_str) = meta.get(&format!("{}actual-size", RESERVED_METADATA_PREFIX_LOWER)) {
if !size_str.is_empty() {
// Todo: deal with error
let size = size_str.parse::<i64>().map_err(|e| std::io::Error::other(e.to_string()))?;
return Ok(size);
}
if let Some(size_str) = self
.user_defined
.get(&format!("{}actual-size", RESERVED_METADATA_PREFIX_LOWER))
{
if !size_str.is_empty() {
// Todo: deal with error
let size = size_str.parse::<i64>().map_err(|e| std::io::Error::other(e.to_string()))?;
return Ok(size);
}
}
let mut actual_size = 0;
self.parts.iter().for_each(|part| {
actual_size += part.actual_size;
@@ -505,7 +502,7 @@ impl ObjectInfo {
let metadata = {
let mut v = fi.metadata.clone();
clean_metadata(&mut v);
Some(v)
v
};
// Convert parts from rustfs_filemeta::ObjectPartInfo to store_api::ObjectPartInfo