Merge pull request #194 from rustfs/fix/metadata

fix: #189 #192

ecstore/src/bucket/metadata.rs

@@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::sync::Arc;
 use time::OffsetDateTime;
-use tracing::{error, warn};
+use tracing::{error, info};
 
 use crate::config::common::{read_config, save_config};
 use crate::error::{Error, Result};
@@ -358,11 +358,12 @@ pub async fn load_bucket_metadata_parse(api: Arc<ECStore>, bucket: &str, parse:
     let mut bm = match read_bucket_metadata(api.clone(), bucket).await {
         Ok(res) => res,
         Err(err) => {
-            warn!("load_bucket_metadata_parse err {:?}", &err);
             if !config::error::is_not_found(&err) {
                 return Err(err);
             }
 
+            info!("bucketmeta {} not found with err {:?}, start to init ", bucket, &err);
+
             BucketMetadata::new(bucket)
         }
     };

ecstore/src/disk/local.rs

@@ -761,7 +761,7 @@ impl LocalDisk {
             Ok(res) => res,
             Err(e) => {
                 if !DiskError::VolumeNotFound.is(&e) && !is_err_file_not_found(&e) {
-                    error!("scan list_dir {}, err {:?}", &current, &e);
+                    warn!("scan list_dir {}, err {:?}", &current, &e);
                 }
 
                 if opts.report_notfound && is_err_file_not_found(&e) && current == &opts.base_dir {
@@ -1443,7 +1443,7 @@ impl DiskAPI for LocalDisk {
         check_path_length(file_path.to_string_lossy().to_string().as_str())?;
 
         //  TODO: writeAllDirect io.copy
-        info!("file_path: {:?}", file_path);
+        // info!("file_path: {:?}", file_path);
         if let Some(parent) = file_path.parent() {
             os::make_dir_all(parent, &volume_dir).await?;
         }

ecstore/src/lib.rs

@@ -28,6 +28,7 @@ pub mod store_api;
 pub mod store_err;
 mod store_init;
 pub mod store_list_objects;
+mod store_utils;
 pub mod utils;
 pub mod xhttp;
 

ecstore/src/set_disk.rs

@@ -3957,6 +3957,8 @@ impl StorageAPI for SetDisks {
         }
         let (fi, _, _) = self.get_object_fileinfo(bucket, object, opts, false).await?;
 
+        // warn!("get object_info fi {:?}", &fi);
+
         let oi = fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended);
 
         Ok(oi)
@@ -4018,14 +4020,14 @@ impl StorageAPI for SetDisks {
         let obj_info = fi.to_object_info(bucket, object, opts.versioned || opts.version_suspended);
 
         if let Some(ref mut metadata) = fi.metadata {
-            for (k, v) in obj_info.user_defined {
+            for (k, v) in obj_info.user_defined.unwrap_or_default() {
                 metadata.insert(k, v);
             }
             fi.metadata = Some(metadata.clone())
         } else {
             let mut metadata = HashMap::new();
 
-            for (k, v) in obj_info.user_defined {
+            for (k, v) in obj_info.user_defined.unwrap_or_default() {
                 metadata.insert(k, v);
             }
 

ecstore/src/store_api.rs

@@ -1,4 +1,5 @@
 use crate::heal::heal_ops::HealSequence;
+use crate::store_utils::clean_metadata;
 use crate::{
     disk::DiskStore,
     error::{Error, Result},
@@ -171,6 +172,7 @@ impl FileInfo {
             version_id = Some(Uuid::nil())
         }
 
+        // etag
         let (content_type, content_encoding, etag) = {
             if let Some(ref meta) = self.metadata {
                 let content_type = meta.get("content-type").cloned();
@@ -182,6 +184,7 @@ impl FileInfo {
                 (None, None, None)
             }
         };
+        // tags
         let user_tags = self
             .metadata
             .as_ref()
@@ -196,6 +199,15 @@ impl FileInfo {
 
         let inlined = self.inline_data();
 
+        // TODO:expires
+        // TODO:ReplicationState
+        // TODO:TransitionedObject
+
+        let metadata = self.metadata.clone().map(|mut v| {
+            clean_metadata(&mut v);
+            v
+        });
+
         ObjectInfo {
             bucket: bucket.to_string(),
             name,
@@ -215,6 +227,7 @@ impl FileInfo {
             successor_mod_time: self.successor_mod_time,
             etag,
             inlined,
+            user_defined: metadata,
             ..Default::default()
         }
     }
@@ -627,7 +640,7 @@ pub struct ObjectInfo {
     // 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 user_defined: Option<HashMap<String, String>>,
     pub parity_blocks: usize,
     pub data_blocks: usize,
     pub version_id: Option<Uuid>,
@@ -646,8 +659,11 @@ pub struct ObjectInfo {
 
 impl ObjectInfo {
     pub fn is_compressed(&self) -> bool {
-        self.user_defined
-            .contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX))
+        if let Some(meta) = &self.user_defined {
+            meta.contains_key(&format!("{}compression", RESERVED_METADATA_PREFIX))
+        } else {
+            false
+        }
     }
 
     pub fn get_actual_size(&self) -> Result<usize> {
@@ -656,13 +672,16 @@ impl ObjectInfo {
         }
 
         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);
+            if let Some(meta) = &self.user_defined {
+                if let Some(size_str) = meta.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;

ecstore/src/store_list_objects.rs

@@ -245,6 +245,8 @@ impl ECStore {
             ..Default::default()
         };
 
+        // warn!("list_objects_generic opts {:?}", &opts);
+
         // use get
         if !opts.prefix.is_empty() && opts.limit == 1 && opts.marker.is_none() {
             match self