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,10 @@ 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_options.pool_max_idle_per_host,
+                        aliyun_opts.http_options.timeout.0,
+                        aliyun_opts.http_options.keep_alive_timeout.0,
+                        aliyun_opts.http_options.keep_alive_interval.0,
                     )
                     .context(OpenObjectStore)?,
                 );
@@ -462,6 +462,24 @@ 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_options.pool_max_idle_per_host,
+                        s3_option.http_options.timeout.0,
+                        s3_option.http_options.keep_alive_timeout.0,
+                        s3_option.http_options.keep_alive_interval.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,7 @@ 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))
-    }
+    pub http_options: HttpOptions,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -119,3 +95,43 @@ 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,
+    pub http_options: HttpOptions,
+}
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct HttpOptions {
+    #[serde(default = "HttpOptions::default_pool_max_idle_per_host")]
+    pub pool_max_idle_per_host: usize,
+    #[serde(default = "HttpOptions::default_timeout")]
+    pub timeout: ReadableDuration,
+    #[serde(default = "HttpOptions::default_keep_alive_time")]
+    pub keep_alive_timeout: ReadableDuration,
+    #[serde(default = "HttpOptions::default_keep_alive_inverval")]
+    pub keep_alive_interval: ReadableDuration,
+}
+
+impl HttpOptions {
+    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))
+    }
+}

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,47 @@
+// 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,
+) -> 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 {
+        // TODO: add to config
+        max_retries: 3,
+        ..Default::default()
+    };
+
+    let endpoint = endpoint.into();
+    let bucket = bucket.into();
+    AmazonS3Builder::new()
+        .with_virtual_hosted_style_request(true)
+        .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()
+}