object_store/
factory.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::{fs, path};
16
17use common_telemetry::info;
18#[cfg(feature = "mysql-object-store")]
19use opendal::services::Mysql;
20use opendal::services::{Fs, Gcs, Oss, S3};
21use snafu::prelude::*;
22
23#[cfg(feature = "mysql-object-store")]
24use crate::config::MysqlConfig;
25use crate::config::{AzblobConfig, FileConfig, GcsConfig, ObjectStoreConfig, OssConfig, S3Config};
26use crate::error::{self, Result};
27use crate::services::Azblob;
28use crate::util::{build_http_context, clean_temp_dir, join_dir, normalize_dir};
29use crate::{ATOMIC_WRITE_DIR, OLD_ATOMIC_WRITE_DIR, ObjectStore, util};
30
31pub async fn new_raw_object_store(
32    store: &ObjectStoreConfig,
33    data_home: &str,
34) -> Result<ObjectStore> {
35    let data_home = normalize_dir(data_home);
36    match store {
37        ObjectStoreConfig::File(file_config) => new_fs_object_store(&data_home, file_config),
38        ObjectStoreConfig::S3(s3_config) => new_s3_object_store(s3_config).await,
39        ObjectStoreConfig::Oss(oss_config) => new_oss_object_store(oss_config).await,
40        ObjectStoreConfig::Azblob(azblob_config) => new_azblob_object_store(azblob_config).await,
41        ObjectStoreConfig::Gcs(gcs_config) => new_gcs_object_store(gcs_config).await,
42        #[cfg(feature = "mysql-object-store")]
43        ObjectStoreConfig::Mysql(mysql_config) => new_mysql_object_store(mysql_config).await,
44    }
45}
46
47#[cfg(feature = "mysql-object-store")]
48pub async fn new_mysql_object_store(mysql_config: &MysqlConfig) -> Result<ObjectStore> {
49    let root = util::normalize_dir(&mysql_config.root);
50    info!(
51        "The mysql object storage table is: {}, root is: {}",
52        mysql_config.table.as_deref().unwrap_or("greptime"),
53        root
54    );
55
56    let builder = Mysql::from(mysql_config);
57    let operator = ObjectStore::new(builder).context(error::InitBackendSnafu)?;
58
59    Ok(operator)
60}
61
62/// A helper function to create a file system object store.
63pub fn new_fs_object_store(data_home: &str, _file_config: &FileConfig) -> Result<ObjectStore> {
64    fs::create_dir_all(path::Path::new(&data_home))
65        .context(error::CreateDirSnafu { dir: data_home })?;
66    info!("The file storage home is: {}", data_home);
67
68    let atomic_write_dir = join_dir(data_home, ATOMIC_WRITE_DIR);
69    clean_temp_dir(&atomic_write_dir)?;
70
71    // Compatible code. Remove this after a major release.
72    let old_atomic_temp_dir = join_dir(data_home, OLD_ATOMIC_WRITE_DIR);
73    clean_temp_dir(&old_atomic_temp_dir)?;
74
75    let builder = Fs::default()
76        .root(data_home)
77        .atomic_write_dir(&atomic_write_dir);
78
79    let object_store = ObjectStore::new(builder).context(error::InitBackendSnafu)?;
80
81    Ok(object_store)
82}
83
84pub async fn new_azblob_object_store(azblob_config: &AzblobConfig) -> Result<ObjectStore> {
85    let root = util::normalize_dir(&azblob_config.connection.root);
86    info!(
87        "The azure storage container is: {}, root is: {}",
88        azblob_config.connection.container, &root
89    );
90
91    let ctx = build_http_context(&azblob_config.http_client)?;
92    let builder = Azblob::from(&azblob_config.connection);
93    let operator = ObjectStore::new(builder)
94        .context(error::InitBackendSnafu)?
95        .with_context(ctx);
96
97    Ok(operator)
98}
99
100pub async fn new_gcs_object_store(gcs_config: &GcsConfig) -> Result<ObjectStore> {
101    let root = util::normalize_dir(&gcs_config.connection.root);
102    info!(
103        "The gcs storage bucket is: {}, root is: {}",
104        gcs_config.connection.bucket, &root
105    );
106
107    let ctx = build_http_context(&gcs_config.http_client)?;
108    let builder = Gcs::from(&gcs_config.connection);
109    let operator = ObjectStore::new(builder)
110        .context(error::InitBackendSnafu)?
111        .with_context(ctx);
112
113    Ok(operator)
114}
115
116pub async fn new_oss_object_store(oss_config: &OssConfig) -> Result<ObjectStore> {
117    let root = util::normalize_dir(&oss_config.connection.root);
118    info!(
119        "The oss storage bucket is: {}, root is: {}",
120        oss_config.connection.bucket, &root
121    );
122
123    let ctx = build_http_context(&oss_config.http_client)?;
124    let builder = Oss::from(&oss_config.connection);
125    let operator = ObjectStore::new(builder)
126        .context(error::InitBackendSnafu)?
127        .with_context(ctx);
128
129    Ok(operator)
130}
131
132pub async fn new_s3_object_store(s3_config: &S3Config) -> Result<ObjectStore> {
133    let root = util::normalize_dir(&s3_config.connection.root);
134    info!(
135        "The s3 storage bucket is: {}, root is: {}",
136        s3_config.connection.bucket, &root
137    );
138
139    let ctx = build_http_context(&s3_config.http_client)?;
140    let builder = S3::from(&s3_config.connection);
141    let operator = ObjectStore::new(builder)
142        .context(error::InitBackendSnafu)?
143        .with_context(ctx);
144
145    Ok(operator)
146}
object_store/factory.rs

object_store/
factory.rs
