feat: expose s3 object store setting

analytic_engine/src/setup.rs

@@ -15,7 +15,7 @@ use object_store::{
     metrics::StoreWithMetrics,
     obkv,
     prefix::StoreWithPrefix,
-    LocalFileSystem, ObjectStoreRef,
+    s3, LocalFileSystem, ObjectStoreRef,
 };
 use snafu::{Backtrace, ResultExt, Snafu};
 use table_engine::engine::{EngineRuntimes, TableEngineRef};
@@ -432,10 +432,12 @@ fn open_storage(
                         aliyun_opts.key_secret,
                         aliyun_opts.endpoint,
                         aliyun_opts.bucket,
-                        aliyun_opts.pool_max_idle_per_host,
-                        aliyun_opts.timeout.0,
-                        aliyun_opts.keep_alive_timeout.0,
-                        aliyun_opts.keep_alive_interval.0,
+                        aliyun_opts.http.pool_max_idle_per_host,
+                        aliyun_opts.http.timeout.0,
+                        aliyun_opts.http.keep_alive_timeout.0,
+                        aliyun_opts.http.keep_alive_interval.0,
+                        aliyun_opts.retry.max_retries,
+                        aliyun_opts.retry.retry_timeout.0,
                     )
                     .context(OpenObjectStore)?,
                 );
@@ -462,6 +464,26 @@ fn open_storage(
                 );
                 Arc::new(StoreWithPrefix::new(obkv_opts.prefix, oss).context(OpenObjectStore)?) as _
             }
+            ObjectStoreOptions::S3(s3_option) => {
+                let oss: ObjectStoreRef = Arc::new(
+                    s3::try_new(
+                        s3_option.region,
+                        s3_option.key_id,
+                        s3_option.key_secret,
+                        s3_option.endpoint,
+                        s3_option.bucket,
+                        s3_option.http.pool_max_idle_per_host,
+                        s3_option.http.timeout.0,
+                        s3_option.http.keep_alive_timeout.0,
+                        s3_option.http.keep_alive_interval.0,
+                        s3_option.retry.max_retries,
+                        s3_option.retry.retry_timeout.0,
+                    )
+                    .context(OpenObjectStore)?,
+                );
+                let store_with_prefix = StoreWithPrefix::new(s3_option.prefix, oss);
+                Arc::new(store_with_prefix.context(OpenObjectStore)?) as _
+            }
         };
 
         if opts.disk_cache_capacity.as_byte() > 0 {

analytic_engine/src/storage_options.rs

@@ -45,6 +45,7 @@ pub enum ObjectStoreOptions {
     Local(LocalOptions),
     Aliyun(AliyunOptions),
     Obkv(ObkvOptions),
+    S3(S3Options),
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -59,32 +60,10 @@ pub struct AliyunOptions {
     pub endpoint: String,
     pub bucket: String,
     pub prefix: String,
-    #[serde(default = "AliyunOptions::default_pool_max_idle_per_host")]
-    pub pool_max_idle_per_host: usize,
-    #[serde(default = "AliyunOptions::default_timeout")]
-    pub timeout: ReadableDuration,
-    #[serde(default = "AliyunOptions::default_keep_alive_time")]
-    pub keep_alive_timeout: ReadableDuration,
-    #[serde(default = "AliyunOptions::default_keep_alive_inverval")]
-    pub keep_alive_interval: ReadableDuration,
-}
-
-impl AliyunOptions {
-    fn default_pool_max_idle_per_host() -> usize {
-        1024
-    }
-
-    fn default_timeout() -> ReadableDuration {
-        ReadableDuration::from(Duration::from_secs(60))
-    }
-
-    fn default_keep_alive_time() -> ReadableDuration {
-        ReadableDuration::from(Duration::from_secs(60))
-    }
-
-    fn default_keep_alive_inverval() -> ReadableDuration {
-        ReadableDuration::from(Duration::from_secs(2))
-    }
+    #[serde(default)]
+    pub http: HttpOptions,
+    #[serde(default)]
+    pub retry: RetryOptions,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -119,3 +98,51 @@ impl ObkvOptions {
         8
     }
 }
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct S3Options {
+    pub region: String,
+    pub key_id: String,
+    pub key_secret: String,
+    pub endpoint: String,
+    pub bucket: String,
+    pub prefix: String,
+    #[serde(default)]
+    pub http: HttpOptions,
+    #[serde(default)]
+    pub retry: RetryOptions,
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HttpOptions {
+    pub pool_max_idle_per_host: usize,
+    pub timeout: ReadableDuration,
+    pub keep_alive_timeout: ReadableDuration,
+    pub keep_alive_interval: ReadableDuration,
+}
+
+impl Default for HttpOptions {
+    fn default() -> Self {
+        Self {
+            pool_max_idle_per_host: 1024,
+            timeout: ReadableDuration::from(Duration::from_secs(60)),
+            keep_alive_timeout: ReadableDuration::from(Duration::from_secs(60)),
+            keep_alive_interval: ReadableDuration::from(Duration::from_secs(2)),
+        }
+    }
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct RetryOptions {
+    pub max_retries: usize,
+    pub retry_timeout: ReadableDuration,
+}
+
+impl Default for RetryOptions {
+    fn default() -> Self {
+        Self {
+            max_retries: 3,
+            retry_timeout: ReadableDuration::from(Duration::from_secs(3 * 60)),
+        }
+    }
+}

components/object_store/src/aliyun.rs

@@ -29,6 +29,8 @@ pub fn try_new(
     timeout: Duration,
     keep_alive_timeout: Duration,
     keep_alive_interval: Duration,
+    max_retries: usize,
+    retry_timeout: Duration,
 ) -> upstream::Result<AmazonS3> {
     let cli_opt = ClientOptions::new()
         .with_allow_http(true)
@@ -38,8 +40,8 @@ pub fn try_new(
         .with_http2_keep_alive_interval(keep_alive_interval)
         .with_timeout(timeout);
     let retry_config = RetryConfig {
-        // TODO: add to config
-        max_retries: 3,
+        max_retries,
+        retry_timeout,
         ..Default::default()
     };
 

components/object_store/src/lib.rs

@@ -16,5 +16,5 @@ pub mod metrics;
 pub mod multipart;
 pub mod obkv;
 pub mod prefix;
-
+pub mod s3;
 pub type ObjectStoreRef = Arc<dyn ObjectStore>;

components/object_store/src/s3.rs

@@ -0,0 +1,48 @@
+// Copyright 2023 CeresDB Project Authors. Licensed under Apache-2.0.
+
+use std::time::Duration;
+
+use upstream::{
+    aws::{AmazonS3, AmazonS3Builder},
+    ClientOptions, RetryConfig,
+};
+
+#[allow(clippy::too_many_arguments)]
+pub fn try_new(
+    region: impl Into<String>,
+    key_id: impl Into<String>,
+    key_secret: impl Into<String>,
+    endpoint: impl Into<String>,
+    bucket: impl Into<String>,
+    pool_max_idle_per_host: impl Into<usize>,
+    timeout: Duration,
+    keep_alive_timeout: Duration,
+    keep_alive_interval: Duration,
+    max_retries: usize,
+    retry_timeout: Duration,
+) -> upstream::Result<AmazonS3> {
+    let cli_opt = ClientOptions::new()
+        .with_allow_http(true)
+        .with_pool_max_idle_per_host(pool_max_idle_per_host.into())
+        .with_http2_keep_alive_timeout(keep_alive_timeout)
+        .with_http2_keep_alive_while_idle()
+        .with_http2_keep_alive_interval(keep_alive_interval)
+        .with_timeout(timeout);
+    let retry_config = RetryConfig {
+        max_retries,
+        retry_timeout,
+        ..Default::default()
+    };
+
+    let endpoint = endpoint.into();
+    let bucket = bucket.into();
+    AmazonS3Builder::new()
+        .with_region(region)
+        .with_access_key_id(key_id)
+        .with_secret_access_key(key_secret)
+        .with_endpoint(endpoint)
+        .with_bucket_name(bucket)
+        .with_client_options(cli_opt)
+        .with_retry(retry_config)
+        .build()
+}