object_store/
util.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::fmt::Display;
16use std::path;
17
18use common_error::root_source;
19use common_telemetry::{debug, error, info, warn};
20use opendal::layers::{
21    LoggingInterceptor, LoggingLayer, RetryEvent, RetryInterceptor, RetryLayer, TracingLayer,
22};
23use opendal::raw::{Operation, ServiceInfo};
24use opendal::services::FS_SCHEME;
25use opendal::{ErrorKind, HttpTransporter, OperationContext};
26use opendal_http_transport_reqwest::ReqwestTransport;
27use snafu::ResultExt;
28
29use crate::config::HttpClientConfig;
30use crate::{ObjectStore, error};
31
32/// Returns true if the object store is not backed by local filesystem.
33pub fn is_object_storage(object_store: &ObjectStore) -> bool {
34    object_store.info().scheme() != FS_SCHEME
35}
36
37/// Join two paths and normalize the output dir.
38///
39/// The output dir is always ends with `/`. e.g.
40/// - `/a/b` join `c` => `/a/b/c/`
41/// - `/a/b` join `/c/` => `/a/b/c/`
42///
43/// All internal `//` will be replaced by `/`.
44pub fn join_dir(parent: &str, child: &str) -> String {
45    // Always adds a `/` to the output path.
46    let output = format!("{parent}/{child}/");
47    normalize_dir(&output)
48}
49
50/// Modified from the `opendal::raw::normalize_root`
51///
52/// # The different
53///
54/// It doesn't always append `/` ahead of the path,
55/// It only keeps `/` ahead if the original path starts with `/`.
56///
57/// Make sure the directory is normalized to style like `abc/def/`.
58///
59/// # Normalize Rules
60///
61/// - All whitespace will be trimmed: ` abc/def ` => `abc/def`
62/// - All leading / will be trimmed: `///abc` => `abc`
63/// - Internal // will be replaced by /: `abc///def` => `abc/def`
64/// - Empty path will be `/`: `` => `/`
65/// - **(Removed❗️)** ~~Add leading `/` if not starts with: `abc/` => `/abc/`~~
66/// - Add trailing `/` if not ends with: `/abc` => `/abc/`
67///
68/// Finally, we will got path like `/path/to/root/`.
69pub fn normalize_dir(v: &str) -> String {
70    let has_root = v.starts_with('/');
71    let mut v = v
72        .split('/')
73        .filter(|v| !v.is_empty())
74        .collect::<Vec<&str>>()
75        .join("/");
76    if has_root {
77        v.insert(0, '/');
78    }
79    if !v.ends_with('/') {
80        v.push('/')
81    }
82    v
83}
84
85/// Push `child` to `parent` dir and normalize the output path.
86///
87/// - Path endswith `/` means it's a dir path.
88/// - Otherwise, it's a file path.
89pub fn join_path(parent: &str, child: &str) -> String {
90    let output = format!("{parent}/{child}");
91    normalize_path(&output)
92}
93
94/// Make sure all operation are constructed by normalized path:
95///
96/// - Path endswith `/` means it's a dir path.
97/// - Otherwise, it's a file path.
98///
99/// # Normalize Rules
100///
101/// - All whitespace will be trimmed: ` abc/def ` => `abc/def`
102/// - Repeated leading / will be trimmed: `///abc` => `/abc`
103/// - Internal // will be replaced by /: `abc///def` => `abc/def`
104/// - Empty path will be `/`: `` => `/`
105pub fn normalize_path(path: &str) -> String {
106    // - all whitespace has been trimmed.
107    let path = path.trim();
108
109    // Fast line for empty path.
110    if path.is_empty() {
111        return "/".to_string();
112    }
113
114    let has_leading = path.starts_with('/');
115    let has_trailing = path.ends_with('/');
116
117    let mut p = path
118        .split('/')
119        .filter(|v| !v.is_empty())
120        .collect::<Vec<_>>()
121        .join("/");
122
123    // If path is not starting with `/` but it should
124    if !p.starts_with('/') && has_leading {
125        p.insert(0, '/');
126    }
127
128    // If path is not ending with `/` but it should
129    if !p.ends_with('/') && has_trailing {
130        p.push('/');
131    }
132
133    p
134}
135
136/// Attaches instrument layers to the object store.
137pub fn with_instrument_layers(object_store: ObjectStore, path_label: bool) -> ObjectStore {
138    object_store
139        .layer(LoggingLayer::new(DefaultLoggingInterceptor))
140        .layer(TracingLayer::new())
141        .layer(crate::layers::build_prometheus_metrics_layer(path_label))
142}
143
144/// Adds retry layer to the object store.
145pub fn with_retry_layers(object_store: ObjectStore) -> ObjectStore {
146    object_store.layer(
147        RetryLayer::new()
148            .with_jitter()
149            .with_notify(PrintDetailedError),
150    )
151}
152
153static LOGGING_TARGET: &str = "opendal::services";
154
155struct LoggingContext<'a>(&'a [(&'a str, &'a str)]);
156
157impl Display for LoggingContext<'_> {
158    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
159        for (i, (k, v)) in self.0.iter().enumerate() {
160            if i > 0 {
161                write!(f, " {}={}", k, v)?;
162            } else {
163                write!(f, "{}={}", k, v)?;
164            }
165        }
166        Ok(())
167    }
168}
169
170#[derive(Debug, Copy, Clone, Default)]
171pub struct DefaultLoggingInterceptor;
172
173impl LoggingInterceptor for DefaultLoggingInterceptor {
174    #[inline]
175    fn log(
176        &self,
177        info: &ServiceInfo,
178        operation: Operation,
179        context: &[(&str, &str)],
180        message: &str,
181        err: Option<&opendal::Error>,
182    ) {
183        if let Some(err) = err {
184            let root = root_source(err);
185            // Print error if it's unexpected, otherwise in error.
186            if err.kind() == ErrorKind::Unexpected {
187                error!(
188                    target: LOGGING_TARGET,
189                    "service={} name={} {}: {operation} {message} {err:#?}, root={root:#?}",
190                    info.scheme(),
191                    info.name(),
192                    LoggingContext(context),
193                );
194            } else {
195                debug!(
196                    target: LOGGING_TARGET,
197                    "service={} name={} {}: {operation} {message} {err}, root={root:?}",
198                    info.scheme(),
199                    info.name(),
200                    LoggingContext(context),
201                );
202            };
203        }
204
205        debug!(
206            target: LOGGING_TARGET,
207            "service={} name={} {}: {operation} {message}",
208            info.scheme(),
209            info.name(),
210            LoggingContext(context),
211        );
212    }
213}
214
215/// Builds an [`OperationContext`] with a custom HTTP transport from `config`.
216pub(crate) fn build_http_context(config: &HttpClientConfig) -> error::Result<OperationContext> {
217    if config.skip_ssl_validation {
218        common_telemetry::warn!(
219            "Skipping SSL validation for object storage HTTP client. Please ensure the environment is trusted."
220        );
221    }
222
223    let client = reqwest::ClientBuilder::new()
224        .pool_max_idle_per_host(config.pool_max_idle_per_host as usize)
225        .connect_timeout(config.connect_timeout)
226        .pool_idle_timeout(config.pool_idle_timeout)
227        .timeout(config.timeout)
228        .danger_accept_invalid_certs(config.skip_ssl_validation)
229        .build()
230        .context(error::BuildHttpClientSnafu)?;
231    let transport = HttpTransporter::new(ReqwestTransport::new(client));
232    Ok(OperationContext::new().with_http_transport(transport))
233}
234
235pub fn clean_temp_dir(dir: &str) -> error::Result<()> {
236    if path::Path::new(&dir).exists() {
237        info!("Begin to clean temp storage directory: {}", dir);
238        std::fs::remove_dir_all(dir).context(error::RemoveDirSnafu { dir })?;
239        info!("Cleaned temp storage directory: {}", dir);
240    }
241
242    Ok(())
243}
244
245/// PrintDetailedError is a retry interceptor that prints error in Debug format in retrying.
246pub struct PrintDetailedError;
247
248// PrintDetailedError is a retry interceptor that prints error in Debug format in retrying.
249impl RetryInterceptor for PrintDetailedError {
250    fn intercept(&self, event: RetryEvent<'_>) {
251        warn!(
252            "Retry after {}s, error: {:#?}",
253            event.retry_after.as_secs_f64(),
254            event.err
255        );
256    }
257}
258
259#[cfg(test)]
260mod tests {
261    use opendal::services::Fs;
262
263    use super::*;
264    use crate::ObjectStore;
265    use crate::util::is_object_storage;
266
267    #[test]
268    fn test_normalize_dir() {
269        assert_eq!("/", normalize_dir("/"));
270        assert_eq!("/", normalize_dir(""));
271        assert_eq!("/test/", normalize_dir("/test"));
272    }
273
274    #[test]
275    fn test_join_dir() {
276        assert_eq!("/", join_dir("", ""));
277        assert_eq!("/", join_dir("/", ""));
278        assert_eq!("/", join_dir("", "/"));
279        assert_eq!("/", join_dir("/", "/"));
280        assert_eq!("/a/", join_dir("/a", ""));
281        assert_eq!("a/b/c/", join_dir("a/b", "c"));
282        assert_eq!("/a/b/c/", join_dir("/a/b", "c"));
283        assert_eq!("/a/b/c/", join_dir("/a/b", "c/"));
284        assert_eq!("/a/b/c/", join_dir("/a/b", "/c/"));
285        assert_eq!("/a/b/c/", join_dir("/a/b", "//c"));
286    }
287
288    #[test]
289    fn test_join_path() {
290        assert_eq!("/", join_path("", ""));
291        assert_eq!("/", join_path("/", ""));
292        assert_eq!("/", join_path("", "/"));
293        assert_eq!("/", join_path("/", "/"));
294        assert_eq!("a/", join_path("a", ""));
295        assert_eq!("/a", join_path("/", "a"));
296        assert_eq!("a/b/c.txt", join_path("a/b", "c.txt"));
297        assert_eq!("/a/b/c.txt", join_path("/a/b", "c.txt"));
298        assert_eq!("/a/b/c/", join_path("/a/b", "c/"));
299        assert_eq!("/a/b/c/", join_path("/a/b", "/c/"));
300        assert_eq!("/a/b/c.txt", join_path("/a/b", "//c.txt"));
301        assert_eq!("abc/def", join_path(" abc", "/def "));
302        assert_eq!("/abc", join_path("//", "/abc"));
303        assert_eq!("abc/def", join_path("abc/", "//def"));
304    }
305
306    #[test]
307    fn test_fs_is_not_object_storage() {
308        let object_store = ObjectStore::new(Fs::default().root("/tmp")).unwrap();
309
310        assert_eq!(FS_SCHEME, object_store.info().scheme());
311        assert!(!is_object_storage(&object_store));
312    }
313}
object_store/util.rs

object_store/
util.rs
