opendal_service_tos/
backend.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use std::fmt::Debug;
19use std::sync::Arc;
20
21use super::reader::*;
22use crate::config::TosConfig;
23use crate::copier::TosCopiers;
24use crate::copier::new_tos_copier;
25use crate::core::constants::X_TOS_VERSION_ID;
26use crate::core::constants::{X_TOS_DIRECTORY, X_TOS_META_PREFIX};
27use crate::core::parse_error;
28use crate::core::tos_parse_into_metadata;
29use crate::core::*;
30use crate::deleter::TosDeleter;
31use crate::lister::{TosLister, TosListers, TosObjectVersionsLister};
32use crate::writer::TosWriter;
33use http::StatusCode;
34use opendal_core::OperationContext;
35use opendal_core::raw::*;
36use opendal_core::{Builder, Capability, EntryMode, Error, ErrorKind, Result};
37use reqsign_core::{Context, OsEnv, ProvideCredentialChain, Signer};
38use reqsign_file_read_tokio::TokioFileRead;
39use reqsign_volcengine_tos::{EnvCredentialProvider, RequestSigner, StaticCredentialProvider};
40
41const TOS_SCHEME: &str = "tos";
42
43/// Builder for Volcengine TOS service.
44#[doc = include_str!("docs.md")]
45#[derive(Default)]
46pub struct TosBuilder {
47    pub(super) config: TosConfig,
48}
49
50impl Debug for TosBuilder {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        f.debug_struct("TosBuilder")
53            .field("config", &self.config)
54            .finish_non_exhaustive()
55    }
56}
57
58impl TosBuilder {
59    /// Set root of this backend.
60    ///
61    /// All operations will happen under this root.
62    pub fn root(mut self, root: &str) -> Self {
63        self.config.root = if root.is_empty() {
64            None
65        } else {
66            Some(root.to_string())
67        };
68        self
69    }
70
71    /// Set bucket name of this backend.
72    pub fn bucket(mut self, bucket: &str) -> Self {
73        self.config.bucket = bucket.to_string();
74        self
75    }
76
77    /// Set endpoint of this backend.
78    ///
79    /// Endpoint must be full uri, e.g.
80    /// - TOS: `https://tos-cn-beijing.volces.com`
81    /// - TOS with region: `https://tos-{region}.volces.com`
82    ///
83    /// If user inputs endpoint without scheme like "tos-cn-beijing.volces.com", we
84    /// will prepend "https://" before it.
85    pub fn endpoint(mut self, endpoint: &str) -> Self {
86        if !endpoint.is_empty() {
87            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string());
88        }
89        self
90    }
91
92    /// Set region of this backend.
93    ///
94    /// Region represent the signing region of this endpoint.
95    ///
96    /// If region is not set, we will try to load it from environment.
97    /// If still not set, default to `cn-beijing`.
98    pub fn region(mut self, region: &str) -> Self {
99        if !region.is_empty() {
100            self.config.region = Some(region.to_string());
101        }
102        self
103    }
104
105    /// Set access_key_id of this backend.
106    ///
107    /// - If access_key_id is set, we will take user's input first.
108    /// - If not, we will try to load it from environment.
109    pub fn access_key_id(mut self, v: &str) -> Self {
110        self.config.access_key_id = Some(v.to_string());
111        self
112    }
113
114    /// Set secret_access_key of this backend.
115    ///
116    /// - If secret_access_key is set, we will take user's input first.
117    /// - If not, we will try to load it from environment.
118    pub fn secret_access_key(mut self, v: &str) -> Self {
119        self.config.secret_access_key = Some(v.to_string());
120        self
121    }
122
123    /// Set security_token of this backend.
124    pub fn security_token(mut self, v: &str) -> Self {
125        self.config.security_token = Some(v.to_string());
126        self
127    }
128
129    /// Skip signature will skip loading credentials and signing requests.
130    pub fn skip_signature(mut self) -> Self {
131        self.config.skip_signature = true;
132        self
133    }
134}
135
136impl Builder for TosBuilder {
137    type Config = TosConfig;
138
139    fn build(self) -> Result<impl Service> {
140        let mut config = self.config;
141        let region = config
142            .region
143            .clone()
144            .unwrap_or_else(|| "cn-beijing".to_string());
145
146        if config.endpoint.is_none() {
147            config.endpoint = Some(format!("https://tos-{}.volces.com", region));
148        }
149
150        let endpoint = config.endpoint.clone().unwrap();
151        let bucket = config.bucket.clone();
152        let root = normalize_root(&config.root.clone().unwrap_or_default());
153
154        let ctx = Context::new().with_file_read(TokioFileRead).with_env(OsEnv);
155
156        let mut provider = ProvideCredentialChain::new().push(EnvCredentialProvider::new());
157
158        if let (Some(ak), Some(sk)) = (&config.access_key_id, &config.secret_access_key) {
159            let static_provider = if let Some(token) = config.security_token.as_deref() {
160                StaticCredentialProvider::new(ak, sk).with_security_token(token)
161            } else {
162                StaticCredentialProvider::new(ak, sk)
163            };
164            provider = provider.push_front(static_provider);
165        }
166
167        let request_signer = RequestSigner::new(&region);
168        let signer = Signer::new(ctx, provider, request_signer);
169
170        let info = ServiceInfo::new(TOS_SCHEME, &root, &bucket);
171        let capability = Capability {
172            stat_with_version: true,
173
174            read: true,
175            read_with_suffix: true,
176            read_with_if_match: true,
177            read_with_if_none_match: true,
178            read_with_if_modified_since: true,
179            read_with_if_unmodified_since: true,
180            read_with_version: true,
181
182            write: true,
183            write_can_empty: true,
184            write_can_multi: true,
185            write_with_cache_control: true,
186            write_with_content_type: true,
187            write_with_content_encoding: true,
188            write_with_if_match: true,
189            write_with_if_not_exists: true,
190            write_with_user_metadata: true,
191            write_multi_min_size: Some(5 * 1024 * 1024),
192            write_multi_max_size: if cfg!(target_pointer_width = "64") {
193                Some(5 * 1024 * 1024 * 1024)
194            } else {
195                Some(usize::MAX)
196            },
197
198            delete: true,
199            delete_max_size: Some(1000),
200            delete_with_version: true,
201
202            copy: true,
203            copy_can_multi: true,
204            copy_multi_min_size: Some(5 * 1024 * 1024),
205            copy_multi_max_size: if cfg!(target_pointer_width = "64") {
206                Some(5 * 1024 * 1024 * 1024)
207            } else {
208                Some(usize::MAX)
209            },
210
211            list: true,
212            list_with_limit: true,
213            list_with_start_after: true,
214            list_with_recursive: true,
215            list_with_versions: true,
216            list_with_deleted: true,
217
218            stat: true,
219            stat_with_if_match: true,
220            stat_with_if_none_match: true,
221            stat_with_if_modified_since: true,
222            stat_with_if_unmodified_since: true,
223
224            shared: true,
225
226            ..Default::default()
227        };
228
229        // Extract domain from endpoint, removing http:// or https:// prefix
230        let endpoint_domain = if let Some(stripped) = endpoint.strip_prefix("http://") {
231            stripped
232        } else if let Some(stripped) = endpoint.strip_prefix("https://") {
233            stripped
234        } else {
235            &endpoint
236        };
237
238        let core = TosCore {
239            info,
240            capability,
241            bucket,
242            endpoint: endpoint.clone(),
243            endpoint_domain: endpoint_domain.to_string(),
244            root,
245            default_storage_class: None,
246            skip_signature: config.skip_signature,
247            signer,
248        };
249
250        Ok(TosBackend {
251            core: Arc::new(core),
252        })
253    }
254}
255
256#[derive(Debug, Clone)]
257pub struct TosBackend {
258    pub(crate) core: Arc<TosCore>,
259}
260
261impl Service for TosBackend {
262    type Reader = oio::StreamReader<TosReader>;
263    type Writer = oio::MultipartWriter<TosWriter>;
264    type Lister = TosListers;
265    type Deleter = oio::BatchDeleter<TosDeleter>;
266    type Copier = TosCopiers;
267
268    fn info(&self) -> ServiceInfo {
269        self.core.info.clone()
270    }
271
272    fn capability(&self) -> Capability {
273        self.core.capability
274    }
275
276    async fn create_dir(
277        &self,
278        _ctx: &OperationContext,
279        _path: &str,
280        _args: OpCreateDir,
281    ) -> Result<RpCreateDir> {
282        Err(Error::new(
283            ErrorKind::Unsupported,
284            "operation is not supported",
285        ))
286    }
287
288    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
289        let resp = self.core.tos_head_object(ctx, path, args).await?;
290
291        let status = resp.status();
292
293        match status {
294            StatusCode::OK => {
295                let headers = resp.headers();
296                let mut meta = tos_parse_into_metadata(path, headers)?;
297
298                let user_meta = parse_prefixed_headers(headers, X_TOS_META_PREFIX);
299                if !user_meta.is_empty() {
300                    meta = meta.with_user_metadata(user_meta);
301                }
302
303                if let Some(v) = parse_header_to_str(headers, X_TOS_VERSION_ID)? {
304                    meta.set_version(v);
305                }
306
307                if let Some(is_dir) = parse_header_to_str(headers, X_TOS_DIRECTORY)?
308                    .map(|v| {
309                        v.parse::<bool>().map_err(|e| {
310                            Error::new(ErrorKind::Unexpected, "header value is not valid integer")
311                                .set_source(e)
312                        })
313                    })
314                    .transpose()?
315                {
316                    meta = meta.with_mode(if is_dir {
317                        EntryMode::DIR
318                    } else {
319                        EntryMode::FILE
320                    });
321                }
322
323                Ok(RpStat::new(meta))
324            }
325            _ => Err(parse_error(resp)),
326        }
327    }
328    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
329        let output: oio::StreamReader<TosReader> = {
330            Ok(oio::StreamReader::new(TosReader::new(
331                self.clone(),
332                ctx.clone(),
333                path,
334                args,
335            )))
336        }?;
337
338        Ok(output)
339    }
340
341    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
342        let output: oio::MultipartWriter<TosWriter> = {
343            let writer = TosWriter::new(self.core.clone(), ctx.clone(), path, args.clone());
344
345            let w = oio::MultipartWriter::new(ctx.executor().clone(), writer, args.concurrent());
346
347            Ok(w)
348        }?;
349
350        Ok(output)
351    }
352
353    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
354        let output: oio::BatchDeleter<TosDeleter> = {
355            let deleter = TosDeleter::new(self.core.clone(), ctx.clone());
356            Ok(oio::BatchDeleter::new(
357                deleter,
358                self.core.capability.delete_max_size,
359            ))
360        }?;
361
362        Ok(output)
363    }
364
365    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
366        let output: TosListers = {
367            let lister = if args.versions() || args.deleted() {
368                TwoWays::Two(oio::PageLister::new(TosObjectVersionsLister::new(
369                    self.core.clone(),
370                    ctx.clone(),
371                    path,
372                    args,
373                )))
374            } else {
375                TwoWays::One(oio::PageLister::new(TosLister::new(
376                    self.core.clone(),
377                    ctx.clone(),
378                    path,
379                    args,
380                )))
381            };
382            Ok(lister)
383        }?;
384
385        Ok(output)
386    }
387
388    fn copy(
389        &self,
390        ctx: &OperationContext,
391        from: &str,
392        to: &str,
393        args: OpCopy,
394        opts: OpCopier,
395    ) -> Result<Self::Copier> {
396        let output: TosCopiers = {
397            let copier = new_tos_copier(self.core.clone(), ctx, from, to, args, opts)?;
398            Ok(copier)
399        }?;
400
401        Ok(output)
402    }
403
404    async fn rename(
405        &self,
406        _ctx: &OperationContext,
407        _from: &str,
408        _to: &str,
409        _args: OpRename,
410    ) -> Result<RpRename> {
411        Err(Error::new(
412            ErrorKind::Unsupported,
413            "operation is not supported",
414        ))
415    }
416
417    async fn presign(
418        &self,
419        _ctx: &OperationContext,
420        _path: &str,
421        _args: OpPresign,
422    ) -> Result<RpPresign> {
423        Err(Error::new(
424            ErrorKind::Unsupported,
425            "operation is not supported",
426        ))
427    }
428}
opendal_service_tos/backend.rs

opendal_service_tos/
backend.rs
