refactor: function args of object store

analytic_engine/src/lib.rs

@@ -15,7 +15,6 @@ mod sampler;
 pub mod setup;
 pub mod space;
 pub mod sst;
-mod storage_options;
 pub mod table;
 pub mod table_options;
 
@@ -26,8 +25,8 @@ pub mod tests;
 use common_util::config::{ReadableDuration, ReadableSize};
 use manifest::details::Options as ManifestOptions;
 use message_queue::kafka::config::Config as KafkaConfig;
+use object_store::config::StorageOptions;
 use serde::{Deserialize, Serialize};
-use storage_options::StorageOptions;
 use table_kv::config::ObkvConfig;
 use wal::{
     message_queue_impl::config::Config as MessageQueueWalConfig,

analytic_engine/src/setup.rs

@@ -10,6 +10,7 @@ use futures::Future;
 use message_queue::kafka::kafka_impl::KafkaImpl;
 use object_store::{
     aliyun,
+    config::{ObjectStoreOptions, StorageOptions},
     disk_cache::DiskCacheStore,
     mem_cache::{MemCache, MemCacheStore},
     metrics::StoreWithMetrics,
@@ -35,7 +36,6 @@ use crate::{
         factory::{FactoryImpl, ObjectStorePicker, ObjectStorePickerRef, ReadFrequency},
         meta_data::cache::{MetaCache, MetaCacheRef},
     },
-    storage_options::{ObjectStoreOptions, StorageOptions},
     Config, ObkvWalConfig, WalStorageConfig,
 };
 
@@ -426,21 +426,8 @@ fn open_storage(
                 Arc::new(store) as _
             }
             ObjectStoreOptions::Aliyun(aliyun_opts) => {
-                let oss: ObjectStoreRef = Arc::new(
-                    aliyun::try_new(
-                        aliyun_opts.key_id,
-                        aliyun_opts.key_secret,
-                        aliyun_opts.endpoint,
-                        aliyun_opts.bucket,
-                        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)?,
-                );
+                let oss: ObjectStoreRef =
+                    Arc::new(aliyun::try_new(&aliyun_opts).context(OpenObjectStore)?);
                 let store_with_prefix = StoreWithPrefix::new(aliyun_opts.prefix, oss);
                 Arc::new(store_with_prefix.context(OpenObjectStore)?) as _
             }
@@ -465,22 +452,8 @@ 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 oss: ObjectStoreRef =
+                    Arc::new(s3::try_new(&s3_option).context(OpenObjectStore)?);
                 let store_with_prefix = StoreWithPrefix::new(s3_option.prefix, oss);
                 Arc::new(store_with_prefix.context(OpenObjectStore)?) as _
             }

analytic_engine/src/tests/util.rs

@@ -17,6 +17,7 @@ use common_util::{
 };
 use futures::stream::StreamExt;
 use log::info;
+use object_store::config::{LocalOptions, ObjectStoreOptions, StorageOptions};
 use table_engine::{
     engine::{
         CreateTableRequest, DropTableRequest, EngineRuntimes, OpenTableRequest,
@@ -31,7 +32,6 @@ use tempfile::TempDir;
 
 use crate::{
     setup::{EngineBuilder, MemWalsOpener, OpenedWals, RocksDBWalsOpener, WalsOpener},
-    storage_options::{LocalOptions, ObjectStoreOptions, StorageOptions},
     tests::table::{self, FixedSchemaTable, RowTuple},
     Config, RocksDBConfig, WalStorageConfig,
 };

components/object_store/src/aliyun.rs

@@ -1,12 +1,12 @@
 // Copyright 2022 CeresDB Project Authors. Licensed under Apache-2.0.
 
-use std::time::Duration;
-
 use upstream::{
     aws::{AmazonS3, AmazonS3Builder},
     ClientOptions, RetryConfig,
 };
 
+use crate::config::AliyunOptions;
+
 fn normalize_endpoint(endpoint: &str, bucket: &str) -> String {
     if endpoint.starts_with("https") {
         format!(
@@ -19,43 +19,31 @@ fn normalize_endpoint(endpoint: &str, bucket: &str) -> String {
     }
 }
 
-#[allow(clippy::too_many_arguments)]
-pub fn try_new(
-    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> {
+pub fn try_new(aliyun_opts: &AliyunOptions) -> 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_pool_max_idle_per_host(aliyun_opts.http.pool_max_idle_per_host)
+        .with_http2_keep_alive_timeout(aliyun_opts.http.keep_alive_timeout.0)
         .with_http2_keep_alive_while_idle()
-        .with_http2_keep_alive_interval(keep_alive_interval)
-        .with_timeout(timeout);
+        .with_http2_keep_alive_interval(aliyun_opts.http.keep_alive_interval.0)
+        .with_timeout(aliyun_opts.http.timeout.0);
     let retry_config = RetryConfig {
-        max_retries,
-        retry_timeout,
+        max_retries: aliyun_opts.retry.max_retries,
+        retry_timeout: aliyun_opts.retry.retry_timeout.0,
         ..Default::default()
     };
 
-    let endpoint = endpoint.into();
-    let bucket = bucket.into();
-    let endpoint = normalize_endpoint(&endpoint, &bucket);
+    let endpoint = &aliyun_opts.endpoint;
+    let bucket = &aliyun_opts.bucket;
+    let endpoint = normalize_endpoint(endpoint, bucket);
     AmazonS3Builder::new()
         .with_virtual_hosted_style_request(true)
         // region is not used when virtual_hosted_style is true,
         // but is required, so dummy is used here
         // https://github.com/apache/arrow-rs/issues/3827
         .with_region("dummy")
-        .with_access_key_id(key_id)
-        .with_secret_access_key(key_secret)
+        .with_access_key_id(&aliyun_opts.key_id)
+        .with_secret_access_key(&aliyun_opts.key_secret)
         .with_endpoint(endpoint)
         .with_bucket_name(bucket)
         .with_client_options(cli_opt)

components/object_store/src/lib.rs

@@ -10,11 +10,13 @@ pub use upstream::{
 };
 
 pub mod aliyun;
+pub mod config;
 pub mod disk_cache;
 pub mod mem_cache;
 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

@@ -1,47 +1,32 @@
 // 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> {
+use crate::config::S3Options;
+
+pub fn try_new(s3_option: &S3Options) -> 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_pool_max_idle_per_host(s3_option.http.pool_max_idle_per_host)
+        .with_http2_keep_alive_timeout(s3_option.http.keep_alive_timeout.0)
         .with_http2_keep_alive_while_idle()
-        .with_http2_keep_alive_interval(keep_alive_interval)
-        .with_timeout(timeout);
+        .with_http2_keep_alive_interval(s3_option.http.keep_alive_interval.0)
+        .with_timeout(s3_option.http.timeout.0);
     let retry_config = RetryConfig {
-        max_retries,
-        retry_timeout,
+        max_retries: s3_option.retry.max_retries,
+        retry_timeout: s3_option.retry.retry_timeout.0,
         ..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_region(&s3_option.region)
+        .with_access_key_id(&s3_option.key_id)
+        .with_secret_access_key(&s3_option.key_secret)
+        .with_endpoint(&s3_option.endpoint)
+        .with_bucket_name(&s3_option.bucket)
         .with_client_options(cli_opt)
         .with_retry(retry_config)
         .build()