Fix/ilm (#721)

* fix tip remote tier error
* fix transitioned_object
* fix filemeta
* add GCS R2
* add aliyun tencent huaweicloud azure gcs r2 backend tier
* fix signer
* change azure to s3
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: loverustfs <155562731+loverustfs@users.noreply.github.com>

crates/ecstore/src/client/api_restore.rs

@@ -18,28 +18,23 @@
 #![allow(unused_must_use)]
 #![allow(clippy::all)]
 
-use bytes::Bytes;
-use http::HeaderMap;
-use std::collections::HashMap;
-use std::io::Cursor;
-use tokio::io::BufReader;
-
 use crate::client::{
     api_error_response::{err_invalid_argument, http_resp_to_error_response},
     api_get_object_acl::AccessControlList,
     api_get_options::GetObjectOptions,
     transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info},
 };
+use bytes::Bytes;
+use http::HeaderMap;
+use s3s::dto::RestoreRequest;
+use std::collections::HashMap;
+use std::io::Cursor;
+use tokio::io::BufReader;
 
 const TIER_STANDARD: &str = "Standard";
 const TIER_BULK: &str = "Bulk";
 const TIER_EXPEDITED: &str = "Expedited";
 
-#[derive(Debug, Default, serde::Serialize)]
-pub struct GlacierJobParameters {
-    pub tier: String,
-}
-
 #[derive(Debug, Default, serde::Serialize, serde::Deserialize)]
 pub struct Encryption {
     pub encryption_type: String,
@@ -65,58 +60,6 @@ pub struct S3 {
     pub user_metadata: MetadataEntry,
 }
 
-#[derive(Debug, Default, serde::Serialize)]
-pub struct SelectParameters {
-    pub expression_type: String,
-    pub expression: String,
-    //input_serialization:  SelectObjectInputSerialization,
-    //output_serialization: SelectObjectOutputSerialization,
-}
-
-#[derive(Debug, Default, serde::Serialize)]
-pub struct OutputLocation(pub S3);
-
-#[derive(Debug, Default, serde::Serialize)]
-pub struct RestoreRequest {
-    pub restore_type: String,
-    pub tier: String,
-    pub days: i64,
-    pub glacier_job_parameters: GlacierJobParameters,
-    pub description: String,
-    pub select_parameters: SelectParameters,
-    pub output_location: OutputLocation,
-}
-
-impl RestoreRequest {
-    fn set_days(&mut self, v: i64) {
-        self.days = v;
-    }
-
-    fn set_glacier_job_parameters(&mut self, v: GlacierJobParameters) {
-        self.glacier_job_parameters = v;
-    }
-
-    fn set_type(&mut self, v: &str) {
-        self.restore_type = v.to_string();
-    }
-
-    fn set_tier(&mut self, v: &str) {
-        self.tier = v.to_string();
-    }
-
-    fn set_description(&mut self, v: &str) {
-        self.description = v.to_string();
-    }
-
-    fn set_select_parameters(&mut self, v: SelectParameters) {
-        self.select_parameters = v;
-    }
-
-    fn set_output_location(&mut self, v: OutputLocation) {
-        self.output_location = v;
-    }
-}
-
 impl TransitionClient {
     pub async fn restore_object(
         &self,
@@ -125,12 +68,13 @@ impl TransitionClient {
         version_id: &str,
         restore_req: &RestoreRequest,
     ) -> Result<(), std::io::Error> {
-        let restore_request = match quick_xml::se::to_string(restore_req) {
+        /*let restore_request = match quick_xml::se::to_string(restore_req) {
             Ok(buf) => buf,
             Err(e) => {
                 return Err(std::io::Error::other(e));
             }
-        };
+        };*/
+        let restore_request = "".to_string();
         let restore_request_bytes = restore_request.as_bytes().to_vec();
 
         let mut url_values = HashMap::new();

crates/ecstore/src/client/bucket_cache.rs

@@ -27,7 +27,7 @@ use tracing::{debug, error, info};
 
 use crate::client::{
     api_error_response::{http_resp_to_error_response, to_error_response},
-    transition_api::{Document, TransitionClient},
+    transition_api::{CreateBucketConfiguration, LocationConstraint, TransitionClient},
 };
 use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
 use s3s::Body;
@@ -82,7 +82,7 @@ impl TransitionClient {
         let req = self.get_bucket_location_request(bucket_name)?;
 
         let mut resp = self.doit(req).await?;
-        location = process_bucket_location_response(resp, bucket_name).await?;
+        location = process_bucket_location_response(resp, bucket_name, &self.tier_type).await?;
         {
             let mut bucket_loc_cache = self.bucket_loc_cache.lock().unwrap();
             bucket_loc_cache.set(bucket_name, &location);
@@ -175,7 +175,11 @@ impl TransitionClient {
     }
 }
 
-async fn process_bucket_location_response(mut resp: http::Response<Body>, bucket_name: &str) -> Result<String, std::io::Error> {
+async fn process_bucket_location_response(
+    mut resp: http::Response<Body>,
+    bucket_name: &str,
+    tier_type: &str,
+) -> Result<String, std::io::Error> {
     //if resp != nil {
     if resp.status() != StatusCode::OK {
         let err_resp = http_resp_to_error_response(&resp, vec![], bucket_name, "");
@@ -209,9 +213,17 @@ async fn process_bucket_location_response(mut resp: http::Response<Body>, bucket
     //}
 
     let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
-    let Document(location_constraint) = quick_xml::de::from_str::<Document>(&String::from_utf8(b).unwrap()).unwrap();
+    let mut location = "".to_string();
+    if tier_type == "huaweicloud" {
+        let d = quick_xml::de::from_str::<CreateBucketConfiguration>(&String::from_utf8(b).unwrap()).unwrap();
+        location = d.location_constraint;
+    } else {
+        if let Ok(LocationConstraint { field }) = quick_xml::de::from_str::<LocationConstraint>(&String::from_utf8(b).unwrap()) {
+            location = field;
+        }
+    }
+    //debug!("location: {}", location);
 
-    let mut location = location_constraint;
     if location == "" {
         location = "us-east-1".to_string();
     }

crates/ecstore/src/client/object_api_utils.rs

@@ -21,6 +21,7 @@
 
 use http::HeaderMap;
 use s3s::dto::ETag;
+use std::pin::Pin;
 use std::{collections::HashMap, io::Cursor, sync::Arc};
 use tokio::io::BufReader;
 
@@ -54,7 +55,7 @@ impl PutObjReader {
     }
 }
 
-pub type ObjReaderFn = Arc<dyn Fn(BufReader<Cursor<Vec<u8>>>, HeaderMap) -> GetObjectReader + 'static>;
+pub type ObjReaderFn<'a> = Arc<dyn Fn(BufReader<Cursor<Vec<u8>>>, HeaderMap) -> GetObjectReader + Send + Sync + 'a>;
 
 fn part_number_to_rangespec(oi: ObjectInfo, part_number: usize) -> Option<HTTPRangeSpec> {
     if oi.size == 0 || oi.parts.len() == 0 {
@@ -108,19 +109,24 @@ fn get_compressed_offsets(oi: ObjectInfo, offset: i64) -> (i64, i64, i64, i64, u
     (compressed_offset, part_skip, first_part_idx, decrypt_skip, seq_num)
 }
 
-pub fn new_getobjectreader(
-    rs: HTTPRangeSpec,
-    oi: &ObjectInfo,
+pub fn new_getobjectreader<'a>(
+    rs: &Option<HTTPRangeSpec>,
+    oi: &'a ObjectInfo,
     opts: &ObjectOptions,
-    h: &HeaderMap,
-) -> Result<(ObjReaderFn, i64, i64), ErrorResponse> {
+    _h: &HeaderMap,
+) -> Result<(ObjReaderFn<'a>, i64, i64), ErrorResponse> {
     //let (_, mut is_encrypted) = crypto.is_encrypted(oi.user_defined)?;
     let mut is_encrypted = false;
     let is_compressed = false; //oi.is_compressed_ok();
 
+    let mut rs_ = None;
+    if rs.is_none() && opts.part_number.is_some() && opts.part_number.unwrap() > 0 {
+        rs_ = part_number_to_rangespec(oi.clone(), opts.part_number.unwrap());
+    }
+
     let mut get_fn: ObjReaderFn;
 
-    let (off, length) = match rs.get_offset_length(oi.size) {
+    let (off, length) = match rs_.unwrap().get_offset_length(oi.size) {
         Ok(x) => x,
         Err(err) => {
             return Err(ErrorResponse {
@@ -136,12 +142,11 @@ pub fn new_getobjectreader(
     };
     get_fn = Arc::new(move |input_reader: BufReader<Cursor<Vec<u8>>>, _: HeaderMap| {
         //Box::pin({
-        /*let r = GetObjectReader {
+        let r = GetObjectReader {
             object_info: oi.clone(),
-            stream: StreamingBlob::new(HashReader::new(input_reader, 10, None, None, 10)),
+            stream: Box::new(input_reader),
         };
-        r*/
-        todo!();
+        r
         //})
     });
 

crates/ecstore/src/client/transition_api.rs

@@ -44,7 +44,7 @@ use std::{
 use time::Duration;
 use time::OffsetDateTime;
 use tokio::io::BufReader;
-use tracing::{debug, error};
+use tracing::{debug, error, warn};
 use url::{Url, form_urlencoded};
 use uuid::Uuid;
 
@@ -109,6 +109,7 @@ pub struct TransitionClient {
     pub health_status: AtomicI32,
     pub trailing_header_support: bool,
     pub max_retries: i64,
+    pub tier_type: String,
 }
 
 #[derive(Debug, Default)]
@@ -132,13 +133,13 @@ pub enum BucketLookupType {
 }
 
 impl TransitionClient {
-    pub async fn new(endpoint: &str, opts: Options) -> Result<TransitionClient, std::io::Error> {
-        let clnt = Self::private_new(endpoint, opts).await?;
+    pub async fn new(endpoint: &str, opts: Options, tier_type: &str) -> Result<TransitionClient, std::io::Error> {
+        let clnt = Self::private_new(endpoint, opts, tier_type).await?;
 
         Ok(clnt)
     }
 
-    async fn private_new(endpoint: &str, opts: Options) -> Result<TransitionClient, std::io::Error> {
+    async fn private_new(endpoint: &str, opts: Options, tier_type: &str) -> Result<TransitionClient, std::io::Error> {
         let endpoint_url = get_endpoint_url(endpoint, opts.secure)?;
 
         //#[cfg(feature = "ring")]
@@ -175,6 +176,7 @@ impl TransitionClient {
             health_status: AtomicI32::new(C_UNKNOWN),
             trailing_header_support: opts.trailing_headers,
             max_retries: opts.max_retries,
+            tier_type: tier_type.to_string(),
         };
 
         {
@@ -283,11 +285,14 @@ impl TransitionClient {
         let mut resp = resp.unwrap();
         debug!("http_resp: {:?}", resp);
 
+        //let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
+        //debug!("http_resp_body: {}", String::from_utf8(b).unwrap());
+
         //if self.is_trace_enabled && !(self.trace_errors_only && resp.status() == StatusCode::OK) {
         if resp.status() != StatusCode::OK {
             //self.dump_http(&cloned_req, &resp)?;
             let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
-            debug!("err_body: {}", String::from_utf8(b).unwrap());
+            warn!("err_body: {}", String::from_utf8(b).unwrap());
         }
 
         Ok(resp)
@@ -330,7 +335,8 @@ impl TransitionClient {
             }
 
             let b = resp.body_mut().store_all_unlimited().await.unwrap().to_vec();
-            let err_response = http_resp_to_error_response(&resp, b.clone(), &metadata.bucket_name, &metadata.object_name);
+            let mut err_response = http_resp_to_error_response(&resp, b.clone(), &metadata.bucket_name, &metadata.object_name);
+            err_response.message = format!("remote tier error: {}", err_response.message);
 
             if self.region == "" {
                 match err_response.code {
@@ -380,9 +386,9 @@ impl TransitionClient {
         method: &http::Method,
         metadata: &mut RequestMetadata,
     ) -> Result<http::Request<Body>, std::io::Error> {
-        let location = metadata.bucket_location.clone();
+        let mut location = metadata.bucket_location.clone();
         if location == "" && metadata.bucket_name != "" {
-            let location = self.get_bucket_location(&metadata.bucket_name).await?;
+            location = self.get_bucket_location(&metadata.bucket_name).await?;
         }
 
         let is_makebucket = metadata.object_name == "" && method == http::Method::PUT && metadata.query_values.len() == 0;
@@ -624,7 +630,7 @@ pub struct TransitionCore(pub Arc<TransitionClient>);
 
 impl TransitionCore {
     pub async fn new(endpoint: &str, opts: Options) -> Result<Self, std::io::Error> {
-        let client = TransitionClient::new(endpoint, opts).await?;
+        let client = TransitionClient::new(endpoint, opts, "").await?;
         Ok(Self(Arc::new(client)))
     }
 
@@ -997,4 +1003,13 @@ impl tower::Service<Request<Body>> for SendRequest {
 }
 
 #[derive(Serialize, Deserialize)]
-pub struct Document(pub String);
+pub struct LocationConstraint {
+    #[serde(rename = "$value")]
+    pub field: String,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct CreateBucketConfiguration {
+    #[serde(rename = "LocationConstraint")]
+    pub location_constraint: String,
+}