object_store/
config.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::time::Duration;
16
17use common_base::readable_size::ReadableSize;
18use common_base::secrets::{ExposeSecret, SecretString};
19#[cfg(feature = "mysql-object-store")]
20use opendal::services::Mysql;
21use opendal::services::{Azblob, Gcs, Oss, S3};
22use serde::{Deserialize, Serialize};
23
24use crate::util;
25
26const DEFAULT_OBJECT_STORE_CACHE_SIZE: ReadableSize = ReadableSize::gb(5);
27
28/// Object storage config
29#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
30#[serde(tag = "type")]
31pub enum ObjectStoreConfig {
32    File(FileConfig),
33    S3(S3Config),
34    Oss(OssConfig),
35    Azblob(AzblobConfig),
36    Gcs(GcsConfig),
37    #[cfg(feature = "mysql-object-store")]
38    Mysql(MysqlConfig),
39}
40
41impl Default for ObjectStoreConfig {
42    fn default() -> Self {
43        ObjectStoreConfig::File(FileConfig {})
44    }
45}
46
47impl ObjectStoreConfig {
48    /// Returns the object storage type name, such as `S3`, `Oss` etc.
49    pub fn provider_name(&self) -> &'static str {
50        match self {
51            Self::File(_) => "File",
52            Self::S3(_) => "S3",
53            Self::Oss(_) => "Oss",
54            Self::Azblob(_) => "Azblob",
55            Self::Gcs(_) => "Gcs",
56            #[cfg(feature = "mysql-object-store")]
57            Self::Mysql(_) => "Mysql",
58        }
59    }
60
61    /// Returns true when it's a remote object storage such as AWS s3 etc.
62    pub fn is_object_storage(&self) -> bool {
63        !matches!(self, Self::File(_))
64    }
65
66    /// Returns the object storage configuration name, return the provider name if it's empty.
67    pub fn config_name(&self) -> &str {
68        let name = match self {
69            // file storage doesn't support name
70            Self::File(_) => self.provider_name(),
71            Self::S3(s3) => &s3.name,
72            Self::Oss(oss) => &oss.name,
73            Self::Azblob(az) => &az.name,
74            Self::Gcs(gcs) => &gcs.name,
75            #[cfg(feature = "mysql-object-store")]
76            Self::Mysql(mysql) => &mysql.name,
77        };
78
79        if name.trim().is_empty() {
80            return self.provider_name();
81        }
82
83        name
84    }
85
86    /// Returns the object storage cache configuration.
87    pub fn cache_config(&self) -> Option<&ObjectStorageCacheConfig> {
88        match self {
89            Self::File(_) => None,
90            Self::S3(s3) => Some(&s3.cache),
91            Self::Oss(oss) => Some(&oss.cache),
92            Self::Azblob(az) => Some(&az.cache),
93            Self::Gcs(gcs) => Some(&gcs.cache),
94            #[cfg(feature = "mysql-object-store")]
95            Self::Mysql(mysql) => Some(&mysql.cache),
96        }
97    }
98
99    /// Returns the mutable object storage cache configuration.
100    pub fn cache_config_mut(&mut self) -> Option<&mut ObjectStorageCacheConfig> {
101        match self {
102            Self::File(_) => None,
103            Self::S3(s3) => Some(&mut s3.cache),
104            Self::Oss(oss) => Some(&mut oss.cache),
105            Self::Azblob(az) => Some(&mut az.cache),
106            Self::Gcs(gcs) => Some(&mut gcs.cache),
107            #[cfg(feature = "mysql-object-store")]
108            Self::Mysql(mysql) => Some(&mut mysql.cache),
109        }
110    }
111}
112
113#[derive(Debug, Clone, Serialize, Default, Deserialize, Eq, PartialEq)]
114#[serde(default)]
115pub struct FileConfig {}
116
117#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
118#[serde(default)]
119pub struct S3Connection {
120    pub bucket: String,
121    pub root: String,
122    #[serde(skip_serializing)]
123    pub access_key_id: SecretString,
124    #[serde(skip_serializing)]
125    pub secret_access_key: SecretString,
126    pub endpoint: Option<String>,
127    pub region: Option<String>,
128    /// Enable virtual host style so that opendal will send API requests in virtual host style instead of path style.
129    /// By default, opendal will send API to https://s3.us-east-1.amazonaws.com/bucket_name
130    /// Enabled, opendal will send API to https://bucket_name.s3.us-east-1.amazonaws.com
131    pub enable_virtual_host_style: bool,
132    /// Disable EC2 metadata service.
133    /// By default, opendal will use EC2 metadata service to load credentials from the instance metadata,
134    /// when access key id and secret access key are not provided.
135    /// If enabled, opendal will *NOT* use EC2 metadata service.
136    pub disable_ec2_metadata: bool,
137}
138
139impl From<&S3Connection> for S3 {
140    fn from(connection: &S3Connection) -> Self {
141        let root = util::normalize_dir(&connection.root);
142
143        let mut builder = S3::default()
144            .root(&root)
145            .bucket(&connection.bucket)
146            .access_key_id(connection.access_key_id.expose_secret())
147            .secret_access_key(connection.secret_access_key.expose_secret());
148
149        if connection.disable_ec2_metadata {
150            builder = builder.disable_ec2_metadata();
151        }
152
153        if let Some(endpoint) = &connection.endpoint {
154            builder = builder.endpoint(endpoint);
155        }
156        if let Some(region) = &connection.region {
157            builder = builder.region(region);
158        }
159        if connection.enable_virtual_host_style {
160            builder = builder.enable_virtual_host_style();
161        }
162
163        builder
164    }
165}
166
167#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
168#[serde(default)]
169pub struct S3Config {
170    pub name: String,
171    #[serde(flatten)]
172    pub connection: S3Connection,
173    #[serde(flatten)]
174    pub cache: ObjectStorageCacheConfig,
175    pub http_client: HttpClientConfig,
176}
177
178#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
179#[serde(default)]
180pub struct OssConnection {
181    pub bucket: String,
182    pub root: String,
183    #[serde(skip_serializing)]
184    pub access_key_id: SecretString,
185    #[serde(skip_serializing)]
186    pub access_key_secret: SecretString,
187    pub endpoint: String,
188}
189
190impl From<&OssConnection> for Oss {
191    fn from(connection: &OssConnection) -> Self {
192        let root = util::normalize_dir(&connection.root);
193        Oss::default()
194            .root(&root)
195            .bucket(&connection.bucket)
196            .endpoint(&connection.endpoint)
197            .access_key_id(connection.access_key_id.expose_secret())
198            .access_key_secret(connection.access_key_secret.expose_secret())
199    }
200}
201
202#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
203#[serde(default)]
204pub struct OssConfig {
205    pub name: String,
206    #[serde(flatten)]
207    pub connection: OssConnection,
208    #[serde(flatten)]
209    pub cache: ObjectStorageCacheConfig,
210    pub http_client: HttpClientConfig,
211}
212
213#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
214#[serde(default)]
215pub struct AzblobConnection {
216    pub container: String,
217    pub root: String,
218    #[serde(skip_serializing)]
219    pub account_name: SecretString,
220    #[serde(skip_serializing)]
221    pub account_key: SecretString,
222    pub endpoint: String,
223    pub sas_token: Option<String>,
224}
225
226impl From<&AzblobConnection> for Azblob {
227    fn from(connection: &AzblobConnection) -> Self {
228        let root = util::normalize_dir(&connection.root);
229        let mut builder = Azblob::default()
230            .root(&root)
231            .container(&connection.container)
232            .endpoint(&connection.endpoint)
233            .account_name(connection.account_name.expose_secret())
234            .account_key(connection.account_key.expose_secret());
235
236        if let Some(token) = &connection.sas_token {
237            builder = builder.sas_token(token);
238        };
239
240        builder
241    }
242}
243
244#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
245#[serde(default)]
246pub struct AzblobConfig {
247    pub name: String,
248    #[serde(flatten)]
249    pub connection: AzblobConnection,
250    #[serde(flatten)]
251    pub cache: ObjectStorageCacheConfig,
252    pub http_client: HttpClientConfig,
253}
254
255#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
256#[serde(default)]
257pub struct GcsConnection {
258    pub root: String,
259    pub bucket: String,
260    pub scope: String,
261    #[serde(skip_serializing)]
262    pub credential_path: SecretString,
263    #[serde(skip_serializing)]
264    pub credential: SecretString,
265    pub endpoint: String,
266}
267
268#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
269#[serde(default)]
270pub struct GcsConfig {
271    pub name: String,
272    #[serde(flatten)]
273    pub connection: GcsConnection,
274    #[serde(flatten)]
275    pub cache: ObjectStorageCacheConfig,
276    pub http_client: HttpClientConfig,
277}
278
279impl From<&GcsConnection> for Gcs {
280    fn from(connection: &GcsConnection) -> Self {
281        let root = util::normalize_dir(&connection.root);
282        Gcs::default()
283            .root(&root)
284            .bucket(&connection.bucket)
285            .scope(&connection.scope)
286            .credential_path(connection.credential_path.expose_secret())
287            .credential(connection.credential.expose_secret())
288            .endpoint(&connection.endpoint)
289    }
290}
291
292#[cfg(feature = "mysql-object-store")]
293#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)]
294#[serde(default)]
295pub struct MysqlConfig {
296    pub name: String,
297    pub root: String,
298    #[serde(skip_serializing)]
299    pub connection_string: SecretString,
300    pub table: Option<String>,
301    #[serde(flatten)]
302    pub cache: ObjectStorageCacheConfig,
303}
304
305#[cfg(feature = "mysql-object-store")]
306impl From<&MysqlConfig> for Mysql {
307    fn from(config: &MysqlConfig) -> Self {
308        let root = util::normalize_dir(&config.root);
309        let mut builder = Mysql::default()
310            .connection_string(config.connection_string.expose_secret())
311            .root(&root)
312            .key_field("key")
313            .value_field("value");
314
315        if let Some(table) = &config.table {
316            builder = builder.table(table);
317        } else {
318            builder = builder.table("greptime");
319        }
320
321        builder
322    }
323}
324
325/// The http client options to the storage.
326#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
327#[serde(default)]
328pub struct HttpClientConfig {
329    /// The maximum idle connection per host allowed in the pool.
330    pub(crate) pool_max_idle_per_host: u32,
331
332    /// The timeout for only the connect phase of a http client.
333    #[serde(with = "humantime_serde")]
334    pub(crate) connect_timeout: Duration,
335
336    /// The total request timeout, applied from when the request starts connecting until the response body has finished.
337    /// Also considered a total deadline.
338    #[serde(with = "humantime_serde")]
339    pub(crate) timeout: Duration,
340
341    /// The timeout for idle sockets being kept-alive.
342    #[serde(with = "humantime_serde")]
343    pub(crate) pool_idle_timeout: Duration,
344
345    /// Skip SSL certificate validation (insecure)
346    pub skip_ssl_validation: bool,
347}
348
349impl Default for HttpClientConfig {
350    fn default() -> Self {
351        Self {
352            pool_max_idle_per_host: 1024,
353            connect_timeout: Duration::from_secs(30),
354            timeout: Duration::from_secs(30),
355            pool_idle_timeout: Duration::from_secs(90),
356            skip_ssl_validation: false,
357        }
358    }
359}
360
361#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
362#[serde(default)]
363pub struct ObjectStorageCacheConfig {
364    /// Whether to enable read cache. If not set, the read cache will be enabled by default.
365    pub enable_read_cache: bool,
366    /// The local file cache directory
367    pub cache_path: String,
368    /// The cache capacity in bytes
369    pub cache_capacity: ReadableSize,
370}
371
372impl Default for ObjectStorageCacheConfig {
373    fn default() -> Self {
374        Self {
375            enable_read_cache: true,
376            // The cache directory is set to the value of data_home in the build_cache_layer process.
377            cache_path: String::default(),
378            cache_capacity: DEFAULT_OBJECT_STORE_CACHE_SIZE,
379        }
380    }
381}
382
383impl ObjectStorageCacheConfig {
384    /// Sanitize the `ObjectStorageCacheConfig` to ensure the config is valid.
385    pub fn sanitize(&mut self, data_home: &str) {
386        // If `cache_path` is unset, default to use `${data_home}` as the local read cache directory.
387        if self.cache_path.is_empty() {
388            self.cache_path = data_home.to_string();
389        }
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use super::*;
396    use crate::config::ObjectStoreConfig;
397
398    #[test]
399    fn test_config_name() {
400        let object_store_config = ObjectStoreConfig::default();
401        assert_eq!("File", object_store_config.config_name());
402
403        let s3_config = ObjectStoreConfig::S3(S3Config::default());
404        assert_eq!("S3", s3_config.config_name());
405        assert_eq!("S3", s3_config.provider_name());
406
407        let s3_config = ObjectStoreConfig::S3(S3Config {
408            name: "test".to_string(),
409            ..Default::default()
410        });
411        assert_eq!("test", s3_config.config_name());
412        assert_eq!("S3", s3_config.provider_name());
413
414        #[cfg(feature = "mysql-object-store")]
415        {
416            let mysql_config = ObjectStoreConfig::Mysql(MysqlConfig::default());
417            assert_eq!("Mysql", mysql_config.config_name());
418            assert_eq!("Mysql", mysql_config.provider_name());
419
420            let mysql_config = ObjectStoreConfig::Mysql(MysqlConfig {
421                name: "test".to_string(),
422                ..Default::default()
423            });
424            assert_eq!("test", mysql_config.config_name());
425            assert_eq!("Mysql", mysql_config.provider_name());
426        }
427    }
428
429    #[test]
430    fn test_is_object_storage() {
431        let store = ObjectStoreConfig::default();
432        assert!(!store.is_object_storage());
433        let s3_config = ObjectStoreConfig::S3(S3Config::default());
434        assert!(s3_config.is_object_storage());
435        let oss_config = ObjectStoreConfig::Oss(OssConfig::default());
436        assert!(oss_config.is_object_storage());
437        let gcs_config = ObjectStoreConfig::Gcs(GcsConfig::default());
438        assert!(gcs_config.is_object_storage());
439        let azblob_config = ObjectStoreConfig::Azblob(AzblobConfig::default());
440        assert!(azblob_config.is_object_storage());
441        #[cfg(feature = "mysql-object-store")]
442        {
443            let mysql_config = ObjectStoreConfig::Mysql(MysqlConfig::default());
444            assert!(mysql_config.is_object_storage());
445        }
446    }
447
448    #[cfg(feature = "mysql-object-store")]
449    #[test]
450    fn test_mysql_config_connection_string_serde() {
451        let config: ObjectStoreConfig = toml::from_str(
452            r#"
453type = "Mysql"
454name = "mysql-store"
455root = "/greptimedb"
456connection_string = "mysql://user:password@127.0.0.1:3306/greptime"
457table = "object_store"
458"#,
459        )
460        .unwrap();
461
462        let ObjectStoreConfig::Mysql(mysql_config) = config else {
463            unreachable!()
464        };
465
466        assert_eq!("mysql-store", mysql_config.name);
467        assert_eq!("/greptimedb", mysql_config.root);
468        assert_eq!(
469            "mysql://user:password@127.0.0.1:3306/greptime",
470            mysql_config.connection_string.expose_secret()
471        );
472        assert_eq!(Some("object_store"), mysql_config.table.as_deref());
473
474        let serialized = toml::to_string(&mysql_config).unwrap();
475        assert!(!serialized.contains("connection_string"));
476        assert!(!serialized.contains("password"));
477    }
478}
object_store/config.rs

object_store/
config.rs
