refactor(services/s3): Migrate to async reqsign (apache#1909)

* refactor(services/s3): Migrate to async reqsign Signed-off-by: Xuanwo <github@xuanwo.io> * Fix test Signed-off-by: Xuanwo <github@xuanwo.io> * Fix test Signed-off-by: Xuanwo <github@xuanwo.io> * Add retry for s3 Signed-off-by: Xuanwo <github@xuanwo.io> * Allow anonymous Signed-off-by: Xuanwo <github@xuanwo.io> * Allow set disable ec2 metadata Signed-off-by: Xuanwo <github@xuanwo.io> * Fix reqsign Signed-off-by: Xuanwo <github@xuanwo.io> --------- Signed-off-by: Xuanwo <github@xuanwo.io>
wcy-fdu · Apr 12, 2023 · 169094d · 169094d
1 parent 2d87542
commit 169094d
Show file tree

Hide file tree

Showing 15 changed files with 1,030 additions and 969 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bindings/nodejs/src/lib.rs b/bindings/nodejs/src/lib.rs
@@ -20,11 +20,11 @@ extern crate napi_derive;
 
 use std::collections::HashMap;
 use std::str::FromStr;
+use std::time::Duration;
 
 use futures::TryStreamExt;
 use napi::bindgen_prelude::*;
 use time::format_description::well_known::Rfc3339;
-use time::Duration;
 
 fn build_operator(
     scheme: opendal::Scheme,
@@ -487,7 +487,7 @@ impl Operator {
     pub async fn presign_read(&self, path: String, expires: u32) -> Result<PresignedRequest> {
         let res = self
             .0
-            .presign_read(&path, Duration::seconds(expires as i64))
+            .presign_read(&path, Duration::from_secs(expires as u64))
             .await
             .map_err(format_napi_error)?;
         Ok(PresignedRequest::new(res))
@@ -510,7 +510,7 @@ impl Operator {
     pub async fn presign_write(&self, path: String, expires: u32) -> Result<PresignedRequest> {
         let res = self
             .0
-            .presign_write(&path, Duration::seconds(expires as i64))
+            .presign_write(&path, Duration::from_secs(expires as u64))
             .await
             .map_err(format_napi_error)?;
         Ok(PresignedRequest::new(res))
@@ -533,7 +533,7 @@ impl Operator {
     pub async fn presign_stat(&self, path: String, expires: u32) -> Result<PresignedRequest> {
         let res = self
             .0
-            .presign_stat(&path, Duration::seconds(expires as i64))
+            .presign_stat(&path, Duration::from_secs(expires as u64))
             .await
             .map_err(format_napi_error)?;
         Ok(PresignedRequest::new(res))

diff --git a/core/Cargo.toml b/core/Cargo.toml
@@ -121,7 +121,7 @@ redis = { version = "0.22", features = [
 ], optional = true }
 # NOTE: we keep this for service migration one by one. And finally we will replace reqsign by v0.9.
 reqsign = "0.8.5"
-reqsign_0_9 = { package = "reqsign", git = "https://github.com/Xuanwo/reqsign", rev = "877292a171bfec9593df27cb4ec94676e77a9d57" }
+reqsign_0_9 = { package = "reqsign", git = "https://github.com/Xuanwo/reqsign", rev = "fde88af3aecf4ba6c39e5d84dc39c5200f8f3a5e" }
 reqwest = { version = "0.11.13", features = [
   "multipart",
   "stream",

diff --git a/core/src/raw/http_util/header.rs b/core/src/raw/http_util/header.rs
@@ -26,6 +26,7 @@ use http::header::ETAG;
 use http::header::LAST_MODIFIED;
 use http::header::LOCATION;
 use http::HeaderMap;
+use http::HeaderValue;
 use md5::Digest;
 use time::format_description::well_known::Rfc2822;
 use time::OffsetDateTime;
@@ -273,6 +274,18 @@ pub fn format_authorization_by_bearer(token: &str) -> Result<String> {
     Ok(format!("Bearer {token}"))
 }
 
+/// Build header value from given string.
+pub fn build_header_value(v: &str) -> Result<HeaderValue> {
+    HeaderValue::from_str(v).map_err(|e| {
+        Error::new(
+            ErrorKind::ConfigInvalid,
+            "header value contains invalid characters",
+        )
+        .with_operation("http_util::build_header_value")
+        .set_source(e)
+    })
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;

diff --git a/core/src/raw/http_util/mod.rs b/core/src/raw/http_util/mod.rs
@@ -30,6 +30,7 @@ pub use body::AsyncBody;
 pub use body::IncomingAsyncBody;
 
 mod header;
+pub use header::build_header_value;
 pub use header::format_authorization_by_basic;
 pub use header::format_authorization_by_bearer;
 pub use header::format_content_md5;

diff --git a/core/src/services/gcs/core.rs b/core/src/services/gcs/core.rs
@@ -19,10 +19,13 @@ use std::fmt::Debug;
 use std::fmt::Formatter;
 use std::fmt::Write;
 
+use backon::ExponentialBuilder;
+use backon::Retryable;
 use http::header::CONTENT_LENGTH;
 use http::header::CONTENT_TYPE;
 use http::Request;
 use http::Response;
+use once_cell::sync::Lazy;
 use reqsign_0_9::GoogleCredentialLoader;
 use reqsign_0_9::GoogleSigner;
 use reqsign_0_9::GoogleToken;
@@ -53,11 +56,13 @@ impl Debug for GcsCore {
     }
 }
 
+static BACKOFF: Lazy<ExponentialBuilder> =
+    Lazy::new(|| ExponentialBuilder::default().with_jitter());
+
 impl GcsCore {
     async fn load_token(&self) -> Result<GoogleToken> {
-        let cred = self
-            .token_loader
-            .load()
+        let cred = { || self.token_loader.load() }
+            .retry(&*BACKOFF)
             .await
             .map_err(new_request_credential_error)?;
 

diff --git a/core/src/services/oss/backend.rs b/core/src/services/oss/backend.rs
@@ -552,7 +552,11 @@ impl Accessor for OssBackend {
         };
 
         self.signer
-            .sign_query(&mut req, args.expire())
+            .sign_query(
+                &mut req,
+                // TODO: convert to std::time::Duration
+                time::Duration::seconds_f64(args.expire().as_secs_f64()),
+            )
             .map_err(new_request_sign_error)?;
 
         // We don't need this request anymore, consume it directly.