From 5644fbf2a6132ca08d8d1296bc5175fcf685949e Mon Sep 17 00:00:00 2001 From: congyi <15605187270@163.com> Date: Tue, 2 Apr 2024 15:30:24 +0800 Subject: [PATCH 1/3] add http client --- Cargo.lock | 1 + src/object_store/Cargo.toml | 1 + src/object_store/src/object/mod.rs | 40 +++++++++++++------ .../src/object/opendal_engine/azblob.rs | 13 +++++- .../src/object/opendal_engine/gcs.rs | 13 +++++- .../src/object/opendal_engine/obs.rs | 12 +++++- .../opendal_engine/opendal_object_store.rs | 20 ++++++++++ .../src/object/opendal_engine/opendal_s3.rs | 5 ++- .../src/object/opendal_engine/oss.rs | 13 +++++- 9 files changed, 97 insertions(+), 21 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c66bb5a91db5..f87ed368c149 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10055,6 +10055,7 @@ dependencies = [ "madsim-tokio", "opendal", "prometheus", + "reqwest", "risingwave_common", "rustls 0.23.4", "spin 0.9.8", diff --git a/src/object_store/Cargo.toml b/src/object_store/Cargo.toml index accdc1ab0965..29144c2cab88 100644 --- a/src/object_store/Cargo.toml +++ b/src/object_store/Cargo.toml @@ -29,6 +29,7 @@ itertools = "0.12" madsim = "0.2.22" opendal = "0.45.1" prometheus = { version = "0.13", features = ["process"] } +reqwest = "0.11" risingwave_common = { workspace = true } rustls = "0.23.4" spin = "0.9" diff --git a/src/object_store/src/object/mod.rs b/src/object_store/src/object/mod.rs index c75447323c30..787d037659d5 100644 --- a/src/object_store/src/object/mod.rs +++ b/src/object_store/src/object/mod.rs @@ -832,18 +832,26 @@ pub async fn build_remote_object_store( let gcs = gcs.strip_prefix("gcs://").unwrap(); let (bucket, root) = gcs.split_once('@').unwrap_or((gcs, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_gcs_engine(bucket.to_string(), root.to_string()) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_gcs_engine( + bucket.to_string(), + root.to_string(), + config.clone(), + ) + .unwrap() + .monitored(metrics, config), ) } obs if obs.starts_with("obs://") => { let obs = obs.strip_prefix("obs://").unwrap(); let (bucket, root) = obs.split_once('@').unwrap_or((obs, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_obs_engine(bucket.to_string(), root.to_string()) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_obs_engine( + bucket.to_string(), + root.to_string(), + config.clone(), + ) + .unwrap() + .monitored(metrics, config), ) } @@ -851,9 +859,13 @@ pub async fn build_remote_object_store( let oss = oss.strip_prefix("oss://").unwrap(); let (bucket, root) = oss.split_once('@').unwrap_or((oss, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_oss_engine(bucket.to_string(), root.to_string()) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_oss_engine( + bucket.to_string(), + root.to_string(), + config.clone(), + ) + .unwrap() + .monitored(metrics, config), ) } webhdfs if webhdfs.starts_with("webhdfs://") => { @@ -869,9 +881,13 @@ pub async fn build_remote_object_store( let azblob = azblob.strip_prefix("azblob://").unwrap(); let (container_name, root) = azblob.split_once('@').unwrap_or((azblob, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_azblob_engine(container_name.to_string(), root.to_string()) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_azblob_engine( + container_name.to_string(), + root.to_string(), + config.clone(), + ) + .unwrap() + .monitored(metrics, config), ) } fs if fs.starts_with("fs://") => { diff --git a/src/object_store/src/object/opendal_engine/azblob.rs b/src/object_store/src/object/opendal_engine/azblob.rs index ae09a0fe2d33..b2cad0675150 100644 --- a/src/object_store/src/object/opendal_engine/azblob.rs +++ b/src/object_store/src/object/opendal_engine/azblob.rs @@ -15,14 +15,19 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Azblob; use opendal::Operator; +use risingwave_common::config::ObjectStoreConfig; -use super::{EngineType, OpendalObjectStore}; +use super::{new_http_client, EngineType, OpendalObjectStore}; use crate::object::ObjectResult; const AZBLOB_ENDPOINT: &str = "AZBLOB_ENDPOINT"; impl OpendalObjectStore { /// create opendal azblob engine. - pub fn new_azblob_engine(container_name: String, root: String) -> ObjectResult { + pub fn new_azblob_engine( + container_name: String, + root: String, + object_store_config: ObjectStoreConfig, + ) -> ObjectResult { // Create azblob backend builder. let mut builder = Azblob::default(); builder.root(&root); @@ -32,6 +37,10 @@ impl OpendalObjectStore { .unwrap_or_else(|_| panic!("AZBLOB_ENDPOINT not found from environment variables")); builder.endpoint(&endpoint); + + let http_client = new_http_client(&object_store_config)?; + builder.http_client(http_client); + let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) diff --git a/src/object_store/src/object/opendal_engine/gcs.rs b/src/object_store/src/object/opendal_engine/gcs.rs index 238db7ff51a0..b849d20546a6 100644 --- a/src/object_store/src/object/opendal_engine/gcs.rs +++ b/src/object_store/src/object/opendal_engine/gcs.rs @@ -15,13 +15,18 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Gcs; use opendal::Operator; +use risingwave_common::config::ObjectStoreConfig; -use super::{EngineType, OpendalObjectStore}; +use super::{new_http_client, EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { /// create opendal gcs engine. - pub fn new_gcs_engine(bucket: String, root: String) -> ObjectResult { + pub fn new_gcs_engine( + bucket: String, + root: String, + object_store_config: ObjectStoreConfig, + ) -> ObjectResult { // Create gcs backend builder. let mut builder = Gcs::default(); @@ -34,6 +39,10 @@ impl OpendalObjectStore { if let Ok(cred) = cred { builder.credential(&cred); } + + let http_client = new_http_client(&object_store_config)?; + builder.http_client(http_client); + let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) diff --git a/src/object_store/src/object/opendal_engine/obs.rs b/src/object_store/src/object/opendal_engine/obs.rs index 6acd79ddbed4..29fa7dcd986a 100644 --- a/src/object_store/src/object/opendal_engine/obs.rs +++ b/src/object_store/src/object/opendal_engine/obs.rs @@ -15,13 +15,18 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Obs; use opendal::Operator; +use risingwave_common::config::ObjectStoreConfig; -use super::{EngineType, OpendalObjectStore}; +use super::{new_http_client, EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { /// create opendal obs engine. - pub fn new_obs_engine(bucket: String, root: String) -> ObjectResult { + pub fn new_obs_engine( + bucket: String, + root: String, + object_store_config: ObjectStoreConfig, + ) -> ObjectResult { // Create obs backend builder. let mut builder = Obs::default(); @@ -40,6 +45,9 @@ impl OpendalObjectStore { builder.endpoint(&endpoint); builder.access_key_id(&access_key_id); builder.secret_access_key(&secret_access_key); + + let http_client = new_http_client(&object_store_config)?; + builder.http_client(http_client); let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) diff --git a/src/object_store/src/object/opendal_engine/opendal_object_store.rs b/src/object_store/src/object/opendal_engine/opendal_object_store.rs index d50422f015c7..18fe066031cb 100644 --- a/src/object_store/src/object/opendal_engine/opendal_object_store.rs +++ b/src/object_store/src/object/opendal_engine/opendal_object_store.rs @@ -13,12 +13,15 @@ // limitations under the License. use std::ops::Range; +use std::time::Duration; use bytes::Bytes; use fail::fail_point; use futures::{stream, StreamExt, TryStreamExt}; +use opendal::raw::HttpClient; use opendal::services::Memory; use opendal::{Metakey, Operator, Writer}; +use risingwave_common::config::ObjectStoreConfig; use risingwave_common::range::RangeBoundsExt; use thiserror_ext::AsReport; @@ -59,6 +62,23 @@ impl OpendalObjectStore { } } +pub fn new_http_client(config: &ObjectStoreConfig) -> ObjectResult { + let mut client_builder = reqwest::ClientBuilder::new(); + + if let Some(keepalive_ms) = config.s3.object_store_keepalive_ms.as_ref() { + client_builder = + client_builder.http2_keep_alive_timeout(Duration::from_millis(*keepalive_ms)); + } + + if let Some(nodelay) = config.s3.object_store_nodelay.as_ref() { + client_builder = client_builder.tcp_nodelay(*nodelay); + } + + client_builder = client_builder.https_only(false); + + Ok(HttpClient::build(client_builder)?) +} + #[async_trait::async_trait] impl ObjectStore for OpendalObjectStore { fn get_object_prefix(&self, obj_id: u64) -> String { diff --git a/src/object_store/src/object/opendal_engine/opendal_s3.rs b/src/object_store/src/object/opendal_engine/opendal_s3.rs index c10aff55d342..a84161826704 100644 --- a/src/object_store/src/object/opendal_engine/opendal_s3.rs +++ b/src/object_store/src/object/opendal_engine/opendal_s3.rs @@ -19,7 +19,7 @@ use opendal::services::S3; use opendal::Operator; use risingwave_common::config::ObjectStoreConfig; -use super::{EngineType, OpendalObjectStore}; +use super::{new_http_client, EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { @@ -40,6 +40,9 @@ impl OpendalObjectStore { builder.enable_virtual_host_style(); } + let http_client = new_http_client(&object_store_config)?; + builder.http_client(http_client); + let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer( diff --git a/src/object_store/src/object/opendal_engine/oss.rs b/src/object_store/src/object/opendal_engine/oss.rs index 0436a4e9b548..09bff832433b 100644 --- a/src/object_store/src/object/opendal_engine/oss.rs +++ b/src/object_store/src/object/opendal_engine/oss.rs @@ -15,13 +15,18 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Oss; use opendal::Operator; +use risingwave_common::config::ObjectStoreConfig; -use super::{EngineType, OpendalObjectStore}; +use super::{new_http_client, EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { /// create opendal oss engine. - pub fn new_oss_engine(bucket: String, root: String) -> ObjectResult { + pub fn new_oss_engine( + bucket: String, + root: String, + object_store_config: ObjectStoreConfig, + ) -> ObjectResult { // Create oss backend builder. let mut builder = Oss::default(); @@ -40,6 +45,10 @@ impl OpendalObjectStore { builder.endpoint(&endpoint); builder.access_key_id(&access_key_id); builder.access_key_secret(&access_key_secret); + + let http_client = new_http_client(&object_store_config)?; + builder.http_client(http_client); + let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) From 8427ecaba35ffe31a9f7aa47280cc5a37345d144 Mon Sep 17 00:00:00 2001 From: congyi <15605187270@163.com> Date: Tue, 2 Apr 2024 15:38:31 +0800 Subject: [PATCH 2/3] tcp keepalive --- .../src/object/opendal_engine/opendal_object_store.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/object_store/src/object/opendal_engine/opendal_object_store.rs b/src/object_store/src/object/opendal_engine/opendal_object_store.rs index 18fe066031cb..ff5afa944243 100644 --- a/src/object_store/src/object/opendal_engine/opendal_object_store.rs +++ b/src/object_store/src/object/opendal_engine/opendal_object_store.rs @@ -66,8 +66,7 @@ pub fn new_http_client(config: &ObjectStoreConfig) -> ObjectResult { let mut client_builder = reqwest::ClientBuilder::new(); if let Some(keepalive_ms) = config.s3.object_store_keepalive_ms.as_ref() { - client_builder = - client_builder.http2_keep_alive_timeout(Duration::from_millis(*keepalive_ms)); + client_builder = client_builder.tcp_keepalive(Duration::from_millis(*keepalive_ms)); } if let Some(nodelay) = config.s3.object_store_nodelay.as_ref() { From b097869f5994a684030e0f669f35909dcad63bc4 Mon Sep 17 00:00:00 2001 From: congyi <15605187270@163.com> Date: Tue, 2 Apr 2024 22:44:32 +0800 Subject: [PATCH 3/3] resolve comments --- src/object_store/src/object/mod.rs | 40 ++++++------------- .../src/object/opendal_engine/azblob.rs | 12 +----- .../src/object/opendal_engine/gcs.rs | 12 +----- .../src/object/opendal_engine/obs.rs | 11 +---- .../opendal_engine/opendal_object_store.rs | 19 --------- .../src/object/opendal_engine/opendal_s3.rs | 19 ++++++++- .../src/object/opendal_engine/oss.rs | 12 +----- 7 files changed, 37 insertions(+), 88 deletions(-) diff --git a/src/object_store/src/object/mod.rs b/src/object_store/src/object/mod.rs index 787d037659d5..c75447323c30 100644 --- a/src/object_store/src/object/mod.rs +++ b/src/object_store/src/object/mod.rs @@ -832,26 +832,18 @@ pub async fn build_remote_object_store( let gcs = gcs.strip_prefix("gcs://").unwrap(); let (bucket, root) = gcs.split_once('@').unwrap_or((gcs, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_gcs_engine( - bucket.to_string(), - root.to_string(), - config.clone(), - ) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_gcs_engine(bucket.to_string(), root.to_string()) + .unwrap() + .monitored(metrics, config), ) } obs if obs.starts_with("obs://") => { let obs = obs.strip_prefix("obs://").unwrap(); let (bucket, root) = obs.split_once('@').unwrap_or((obs, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_obs_engine( - bucket.to_string(), - root.to_string(), - config.clone(), - ) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_obs_engine(bucket.to_string(), root.to_string()) + .unwrap() + .monitored(metrics, config), ) } @@ -859,13 +851,9 @@ pub async fn build_remote_object_store( let oss = oss.strip_prefix("oss://").unwrap(); let (bucket, root) = oss.split_once('@').unwrap_or((oss, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_oss_engine( - bucket.to_string(), - root.to_string(), - config.clone(), - ) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_oss_engine(bucket.to_string(), root.to_string()) + .unwrap() + .monitored(metrics, config), ) } webhdfs if webhdfs.starts_with("webhdfs://") => { @@ -881,13 +869,9 @@ pub async fn build_remote_object_store( let azblob = azblob.strip_prefix("azblob://").unwrap(); let (container_name, root) = azblob.split_once('@').unwrap_or((azblob, "")); ObjectStoreImpl::Opendal( - OpendalObjectStore::new_azblob_engine( - container_name.to_string(), - root.to_string(), - config.clone(), - ) - .unwrap() - .monitored(metrics, config), + OpendalObjectStore::new_azblob_engine(container_name.to_string(), root.to_string()) + .unwrap() + .monitored(metrics, config), ) } fs if fs.starts_with("fs://") => { diff --git a/src/object_store/src/object/opendal_engine/azblob.rs b/src/object_store/src/object/opendal_engine/azblob.rs index b2cad0675150..8ac8caf09fda 100644 --- a/src/object_store/src/object/opendal_engine/azblob.rs +++ b/src/object_store/src/object/opendal_engine/azblob.rs @@ -15,19 +15,14 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Azblob; use opendal::Operator; -use risingwave_common::config::ObjectStoreConfig; -use super::{new_http_client, EngineType, OpendalObjectStore}; +use super::{EngineType, OpendalObjectStore}; use crate::object::ObjectResult; const AZBLOB_ENDPOINT: &str = "AZBLOB_ENDPOINT"; impl OpendalObjectStore { /// create opendal azblob engine. - pub fn new_azblob_engine( - container_name: String, - root: String, - object_store_config: ObjectStoreConfig, - ) -> ObjectResult { + pub fn new_azblob_engine(container_name: String, root: String) -> ObjectResult { // Create azblob backend builder. let mut builder = Azblob::default(); builder.root(&root); @@ -38,9 +33,6 @@ impl OpendalObjectStore { builder.endpoint(&endpoint); - let http_client = new_http_client(&object_store_config)?; - builder.http_client(http_client); - let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) diff --git a/src/object_store/src/object/opendal_engine/gcs.rs b/src/object_store/src/object/opendal_engine/gcs.rs index b849d20546a6..0577288005fc 100644 --- a/src/object_store/src/object/opendal_engine/gcs.rs +++ b/src/object_store/src/object/opendal_engine/gcs.rs @@ -15,18 +15,13 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Gcs; use opendal::Operator; -use risingwave_common::config::ObjectStoreConfig; -use super::{new_http_client, EngineType, OpendalObjectStore}; +use super::{EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { /// create opendal gcs engine. - pub fn new_gcs_engine( - bucket: String, - root: String, - object_store_config: ObjectStoreConfig, - ) -> ObjectResult { + pub fn new_gcs_engine(bucket: String, root: String) -> ObjectResult { // Create gcs backend builder. let mut builder = Gcs::default(); @@ -40,9 +35,6 @@ impl OpendalObjectStore { builder.credential(&cred); } - let http_client = new_http_client(&object_store_config)?; - builder.http_client(http_client); - let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) diff --git a/src/object_store/src/object/opendal_engine/obs.rs b/src/object_store/src/object/opendal_engine/obs.rs index 29fa7dcd986a..4ddf9579685f 100644 --- a/src/object_store/src/object/opendal_engine/obs.rs +++ b/src/object_store/src/object/opendal_engine/obs.rs @@ -15,18 +15,13 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Obs; use opendal::Operator; -use risingwave_common::config::ObjectStoreConfig; -use super::{new_http_client, EngineType, OpendalObjectStore}; +use super::{EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { /// create opendal obs engine. - pub fn new_obs_engine( - bucket: String, - root: String, - object_store_config: ObjectStoreConfig, - ) -> ObjectResult { + pub fn new_obs_engine(bucket: String, root: String) -> ObjectResult { // Create obs backend builder. let mut builder = Obs::default(); @@ -46,8 +41,6 @@ impl OpendalObjectStore { builder.access_key_id(&access_key_id); builder.secret_access_key(&secret_access_key); - let http_client = new_http_client(&object_store_config)?; - builder.http_client(http_client); let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default()) diff --git a/src/object_store/src/object/opendal_engine/opendal_object_store.rs b/src/object_store/src/object/opendal_engine/opendal_object_store.rs index ff5afa944243..d50422f015c7 100644 --- a/src/object_store/src/object/opendal_engine/opendal_object_store.rs +++ b/src/object_store/src/object/opendal_engine/opendal_object_store.rs @@ -13,15 +13,12 @@ // limitations under the License. use std::ops::Range; -use std::time::Duration; use bytes::Bytes; use fail::fail_point; use futures::{stream, StreamExt, TryStreamExt}; -use opendal::raw::HttpClient; use opendal::services::Memory; use opendal::{Metakey, Operator, Writer}; -use risingwave_common::config::ObjectStoreConfig; use risingwave_common::range::RangeBoundsExt; use thiserror_ext::AsReport; @@ -62,22 +59,6 @@ impl OpendalObjectStore { } } -pub fn new_http_client(config: &ObjectStoreConfig) -> ObjectResult { - let mut client_builder = reqwest::ClientBuilder::new(); - - if let Some(keepalive_ms) = config.s3.object_store_keepalive_ms.as_ref() { - client_builder = client_builder.tcp_keepalive(Duration::from_millis(*keepalive_ms)); - } - - if let Some(nodelay) = config.s3.object_store_nodelay.as_ref() { - client_builder = client_builder.tcp_nodelay(*nodelay); - } - - client_builder = client_builder.https_only(false); - - Ok(HttpClient::build(client_builder)?) -} - #[async_trait::async_trait] impl ObjectStore for OpendalObjectStore { fn get_object_prefix(&self, obj_id: u64) -> String { diff --git a/src/object_store/src/object/opendal_engine/opendal_s3.rs b/src/object_store/src/object/opendal_engine/opendal_s3.rs index a84161826704..db2c7732d8fb 100644 --- a/src/object_store/src/object/opendal_engine/opendal_s3.rs +++ b/src/object_store/src/object/opendal_engine/opendal_s3.rs @@ -15,11 +15,12 @@ use std::time::Duration; use opendal::layers::{LoggingLayer, RetryLayer}; +use opendal::raw::HttpClient; use opendal::services::S3; use opendal::Operator; use risingwave_common::config::ObjectStoreConfig; -use super::{new_http_client, EngineType, OpendalObjectStore}; +use super::{EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { @@ -40,7 +41,7 @@ impl OpendalObjectStore { builder.enable_virtual_host_style(); } - let http_client = new_http_client(&object_store_config)?; + let http_client = Self::new_http_client(&object_store_config)?; builder.http_client(http_client); let op: Operator = Operator::new(builder)? @@ -63,4 +64,18 @@ impl OpendalObjectStore { engine_type: EngineType::S3, }) } + + pub fn new_http_client(config: &ObjectStoreConfig) -> ObjectResult { + let mut client_builder = reqwest::ClientBuilder::new(); + + if let Some(keepalive_ms) = config.s3.object_store_keepalive_ms.as_ref() { + client_builder = client_builder.tcp_keepalive(Duration::from_millis(*keepalive_ms)); + } + + if let Some(nodelay) = config.s3.object_store_nodelay.as_ref() { + client_builder = client_builder.tcp_nodelay(*nodelay); + } + + Ok(HttpClient::build(client_builder)?) + } } diff --git a/src/object_store/src/object/opendal_engine/oss.rs b/src/object_store/src/object/opendal_engine/oss.rs index 09bff832433b..e215b6f93d31 100644 --- a/src/object_store/src/object/opendal_engine/oss.rs +++ b/src/object_store/src/object/opendal_engine/oss.rs @@ -15,18 +15,13 @@ use opendal::layers::{LoggingLayer, RetryLayer}; use opendal::services::Oss; use opendal::Operator; -use risingwave_common::config::ObjectStoreConfig; -use super::{new_http_client, EngineType, OpendalObjectStore}; +use super::{EngineType, OpendalObjectStore}; use crate::object::ObjectResult; impl OpendalObjectStore { /// create opendal oss engine. - pub fn new_oss_engine( - bucket: String, - root: String, - object_store_config: ObjectStoreConfig, - ) -> ObjectResult { + pub fn new_oss_engine(bucket: String, root: String) -> ObjectResult { // Create oss backend builder. let mut builder = Oss::default(); @@ -46,9 +41,6 @@ impl OpendalObjectStore { builder.access_key_id(&access_key_id); builder.access_key_secret(&access_key_secret); - let http_client = new_http_client(&object_store_config)?; - builder.http_client(http_client); - let op: Operator = Operator::new(builder)? .layer(LoggingLayer::default()) .layer(RetryLayer::default())