From 49160f03c0df04f4d5a531041adba502d7aa225d Mon Sep 17 00:00:00 2001 From: weisd Date: Fri, 8 Nov 2024 17:46:19 +0800 Subject: [PATCH] fix content-type --- ecstore/src/options.rs | 69 ++++++++++++++++++++++++++++++++++---- ecstore/src/store_api.rs | 2 ++ ecstore/src/xhttp.rs | 3 +- rustfs/src/storage/ecfs.rs | 66 ++++++++++++++++++++++++++++++------ 4 files changed, 122 insertions(+), 18 deletions(-) diff --git a/ecstore/src/options.rs b/ecstore/src/options.rs index 098046b95..8664afd0b 100644 --- a/ecstore/src/options.rs +++ b/ecstore/src/options.rs @@ -1,12 +1,14 @@ -use http::{HeaderMap, HeaderValue}; -use uuid::Uuid; - +use crate::bucket::metadata; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::error::{Error, Result}; use crate::store_api::ObjectOptions; use crate::store_err::StorageError; use crate::utils::path::is_dir_object; +use http::{HeaderMap, HeaderValue}; +use lazy_static::lazy_static; use std::collections::HashMap; +use tracing::warn; +use uuid::Uuid; pub async fn put_opts( bucket: &str, @@ -58,14 +60,67 @@ pub fn put_opts_from_headers( headers: &HeaderMap, metadata: Option>, ) -> Result { - // TODO custom headers + let metadata = metadata.unwrap_or_default(); + get_default_opts(headers, metadata, false) } fn get_default_opts( - _headers: &HeaderMap, - _metadata: Option>, + headers: &HeaderMap, + metadata: HashMap, _copy_source: bool, ) -> Result { - Ok(ObjectOptions::default()) + warn!("get headers: {:?}", &headers); + warn!("get metadata: {:?}", &metadata); + + Ok(ObjectOptions { + user_defined: metadata.clone(), + ..Default::default() + }) +} + +pub fn extract_metadata(headers: &HeaderMap) -> HashMap { + let mut metadata = HashMap::new(); + + extract_metadata_from_mime(&headers, &mut metadata); + + metadata +} + +fn extract_metadata_from_mime(headers: &HeaderMap, metadata: &mut HashMap) { + for (k, v) in headers.iter() { + if k.as_str().starts_with("x-amz-meta-") { + metadata.insert(k.to_string(), String::from_utf8_lossy(v.as_bytes()).to_string()); + continue; + } + + if k.as_str().starts_with("x-rustfs-meta-") { + metadata.insert(k.to_string(), String::from_utf8_lossy(v.as_bytes()).to_string()); + continue; + } + + for hd in SUPPORTED_HEADERS.iter() { + if k.as_str() == *hd { + metadata.insert(k.to_string(), String::from_utf8_lossy(v.as_bytes()).to_string()); + continue; + } + } + } + + if !metadata.contains_key("content-type") { + metadata.insert("content-type".to_owned(), "binary/octet-stream".to_owned()); + } +} +lazy_static! { + static ref SUPPORTED_HEADERS: Vec<&'static str> = vec![ + "content-type", + "cache-control", + "content-language", + "content-encoding", + "content-disposition", + "x-amz-storage-class", + "x-amz-tagging", + "expires", + "x-amz-replication-status" + ]; } diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 21df384ed..8cc0342f6 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -537,6 +537,8 @@ pub struct ObjectOptions { pub data_movement: bool, pub src_pool_idx: usize, + pub user_defined: HashMap, + pub preserve_etag: Option, } // impl Default for ObjectOptions { diff --git a/ecstore/src/xhttp.rs b/ecstore/src/xhttp.rs index f0e2a15b1..b4a3a2b11 100644 --- a/ecstore/src/xhttp.rs +++ b/ecstore/src/xhttp.rs @@ -1 +1,2 @@ -pub(crate) const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging"; +pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging"; +pub const AMZ_STORAGE_CLASS: &str = "x-amz-storage-class"; diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index a98ef7a24..51fd48968 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -18,6 +18,7 @@ use ecstore::bucket::tagging::encode_tags; use ecstore::bucket::versioning_sys::BucketVersioningSys; use ecstore::disk::error::DiskError; use ecstore::new_object_layer_fn; +use ecstore::options::extract_metadata; use ecstore::options::put_opts; use ecstore::store_api::BucketOptions; use ecstore::store_api::CompletePart; @@ -30,6 +31,7 @@ use ecstore::store_api::ObjectOptions; use ecstore::store_api::ObjectToDelete; use ecstore::store_api::PutObjReader; use ecstore::store_api::StorageAPI; +use ecstore::xhttp; use futures::pin_mut; use futures::{Stream, StreamExt}; use http::HeaderMap; @@ -45,6 +47,7 @@ use s3s::{S3Request, S3Response}; use std::fmt::Debug; use std::str::FromStr; use tracing::debug; +use tracing::error; use tracing::info; use transform_stream::AsyncTryStream; use uuid::Uuid; @@ -325,14 +328,28 @@ impl S3 for FS { let info = reader.object_info; - let content_type = try_!(ContentType::from_str("application/x-msdownload")); + let content_type = { + if let Some(content_type) = info.content_type { + let ct = match ContentType::from_str(&content_type) { + Ok(res) => Some(res), + Err(err) => { + error!("parse content-type err {} {:?}", &content_type, err); + // + None + } + }; + ct + } else { + None + } + }; let last_modified = info.mod_time.map(Timestamp::from); let output = GetObjectOutput { body: Some(reader.stream), content_length: Some(info.size as i64), last_modified, - content_type: Some(content_type), + content_type, ..Default::default() }; @@ -378,12 +395,26 @@ impl S3 for FS { let info = try_!(store.get_object_info(&bucket, &key, &ObjectOptions::default()).await); debug!("info {:?}", info); - let content_type = try_!(ContentType::from_str("application/x-msdownload")); + let content_type = { + if let Some(content_type) = info.content_type { + let ct = match ContentType::from_str(&content_type) { + Ok(res) => Some(res), + Err(err) => { + error!("parse content-type err {} {:?}", &content_type, err); + // + None + } + }; + ct + } else { + None + } + }; let last_modified = info.mod_time.map(Timestamp::from); let output = HeadObjectOutput { content_length: Some(try_!(i64::try_from(info.size))), - content_type: Some(content_type), + content_type, last_modified, // metadata: object_metadata, ..Default::default() @@ -534,8 +565,8 @@ impl S3 for FS { body, bucket, key, - metadata, content_length, + tagging, .. } = input; @@ -552,7 +583,12 @@ impl S3 for FS { None => return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())), }; - let opts: ObjectOptions = try_!(put_opts(&bucket, &key, None, &req.headers, metadata).await); + let mut metadata = extract_metadata(&req.headers); + if let Some(tags) = tagging { + metadata.insert(xhttp::AMZ_OBJECT_TAGGING.to_owned(), tags); + } + + let opts: ObjectOptions = try_!(put_opts(&bucket, &key, None, &req.headers, Some(metadata)).await); let obj_info = try_!(store.put_object(&bucket, &key, &mut reader, &opts).await); @@ -573,12 +609,12 @@ impl S3 for FS { req: S3Request, ) -> S3Result> { let CreateMultipartUploadInput { - bucket, key, metadata, .. + bucket, key, tagging, .. } = req.input; // mc cp step 3 - debug!("create_multipart_upload meta {:?}", &metadata); + // debug!("create_multipart_upload meta {:?}", &metadata); let layer = new_object_layer_fn(); let lock = layer.read().await; @@ -587,8 +623,15 @@ impl S3 for FS { None => return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())), }; - let MultipartUploadResult { upload_id, .. } = - try_!(store.new_multipart_upload(&bucket, &key, &ObjectOptions::default()).await); + let mut metadata = extract_metadata(&req.headers); + + if let Some(tags) = tagging { + metadata.insert(xhttp::AMZ_OBJECT_TAGGING.to_owned(), tags); + } + + let opts: ObjectOptions = try_!(put_opts(&bucket, &key, None, &req.headers, Some(metadata)).await); + + let MultipartUploadResult { upload_id, .. } = try_!(store.new_multipart_upload(&bucket, &key, &opts).await); let output = CreateMultipartUploadOutput { bucket: Some(bucket), @@ -609,6 +652,7 @@ impl S3 for FS { upload_id, part_number, content_length, + // content_md5, .. } = req.input; @@ -630,6 +674,8 @@ impl S3 for FS { None => return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())), }; + // TODO: hash_reader + let info = try_!( store .put_object_part(&bucket, &key, &upload_id, part_id, &mut data, &opts)