opendal_service_azdls/
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 http::StatusCode;
22use log::debug;
23use reqsign_azure_storage::Credential;
24use reqsign_azure_storage::DefaultCredentialProvider;
25use reqsign_azure_storage::RequestSigner;
26use reqsign_azure_storage::StaticCredentialProvider;
27use reqsign_core::Context;
28use reqsign_core::Env as _;
29use reqsign_core::OsEnv;
30use reqsign_core::ProvideCredentialChain;
31use reqsign_core::Signer;
32use reqsign_core::StaticEnv;
33use reqsign_file_read_tokio::TokioFileRead;
34
35use super::AZDLS_SCHEME;
36use super::config::AzdlsConfig;
37use super::core::DIRECTORY;
38use super::core::parse_error;
39use super::core::{AzdlsCore, ErrorContext};
40use super::deleter::AzdlsDeleter;
41use super::lister::AzdlsLister;
42use super::reader::*;
43use super::writer::AzdlsLazyPositionWriter;
44use super::writer::AzdlsWriter;
45use super::writer::AzdlsWriters;
46use opendal_core::raw::*;
47use opendal_core::*;
48use opendal_service_azure_common::{
49    AzureStorageConfig as AzureConnectionConfig, AzureStorageService,
50    azure_account_name_from_endpoint, azure_config_from_connection_string,
51};
52
53impl From<AzureConnectionConfig> for AzdlsConfig {
54    fn from(config: AzureConnectionConfig) -> Self {
55        AzdlsConfig {
56            endpoint: config.endpoint,
57            account_name: config.account_name,
58            account_key: config.account_key,
59            client_secret: config.client_secret,
60            tenant_id: config.tenant_id,
61            client_id: config.client_id,
62            sas_token: config.sas_token,
63            authority_host: config.authority_host,
64            ..Default::default()
65        }
66    }
67}
68
69/// Azure Data Lake Storage Gen2 Support.
70#[doc = include_str!("docs.md")]
71#[derive(Default)]
72pub struct AzdlsBuilder {
73    pub(super) config: AzdlsConfig,
74    pub(super) credential_providers: Option<ProvideCredentialChain<Credential>>,
75}
76
77impl Debug for AzdlsBuilder {
78    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
79        f.debug_struct("AzdlsBuilder")
80            .field("config", &self.config)
81            .finish_non_exhaustive()
82    }
83}
84
85impl AzdlsBuilder {
86    /// Set root of this backend.
87    ///
88    /// All operations will happen under this root.
89    pub fn root(mut self, root: &str) -> Self {
90        self.config.root = if root.is_empty() {
91            None
92        } else {
93            Some(root.to_string())
94        };
95
96        self
97    }
98
99    /// Set filesystem name of this backend.
100    pub fn filesystem(mut self, filesystem: &str) -> Self {
101        self.config.filesystem = filesystem.to_string();
102
103        self
104    }
105
106    /// Set endpoint of this backend.
107    ///
108    /// Endpoint must be full uri, e.g.
109    ///
110    /// - Azblob: `https://accountname.blob.core.windows.net`
111    /// - Azurite: `http://127.0.0.1:10000/devstoreaccount1`
112    pub fn endpoint(mut self, endpoint: &str) -> Self {
113        if !endpoint.is_empty() {
114            // Trim trailing `/` so that we can accept `http://127.0.0.1:9000/`
115            self.config.endpoint = Some(endpoint.trim_end_matches('/').to_string());
116        }
117
118        self
119    }
120
121    /// Set account_name of this backend.
122    ///
123    /// - If account_name is set, we will take user's input first.
124    /// - If not, we will try to load it from environment.
125    pub fn account_name(mut self, account_name: &str) -> Self {
126        if !account_name.is_empty() {
127            self.config.account_name = Some(account_name.to_string());
128        }
129
130        self
131    }
132
133    /// Set account_key of this backend.
134    ///
135    /// - If account_key is set, we will take user's input first.
136    /// - If not, we will try to load it from environment.
137    pub fn account_key(mut self, account_key: &str) -> Self {
138        if !account_key.is_empty() {
139            self.config.account_key = Some(account_key.to_string());
140        }
141
142        self
143    }
144
145    /// Set client_secret of this backend.
146    ///
147    /// - If client_secret is set, we will take user's input first.
148    /// - If not, we will try to load it from environment.
149    /// - required for client_credentials authentication
150    pub fn client_secret(mut self, client_secret: &str) -> Self {
151        if !client_secret.is_empty() {
152            self.config.client_secret = Some(client_secret.to_string());
153        }
154
155        self
156    }
157
158    /// Set tenant_id of this backend.
159    ///
160    /// - If tenant_id is set, we will take user's input first.
161    /// - If not, we will try to load it from environment.
162    /// - required for client_credentials authentication
163    pub fn tenant_id(mut self, tenant_id: &str) -> Self {
164        if !tenant_id.is_empty() {
165            self.config.tenant_id = Some(tenant_id.to_string());
166        }
167
168        self
169    }
170
171    /// Set client_id of this backend.
172    ///
173    /// - If client_id is set, we will take user's input first.
174    /// - If not, we will try to load it from environment.
175    /// - required for client_credentials authentication
176    pub fn client_id(mut self, client_id: &str) -> Self {
177        if !client_id.is_empty() {
178            self.config.client_id = Some(client_id.to_string());
179        }
180
181        self
182    }
183
184    /// Set the sas_token of this backend.
185    pub fn sas_token(mut self, sas_token: &str) -> Self {
186        if !sas_token.is_empty() {
187            self.config.sas_token = Some(sas_token.to_string());
188        }
189
190        self
191    }
192
193    /// Replace the credential providers with a custom chain.
194    pub fn credential_provider_chain(mut self, chain: ProvideCredentialChain<Credential>) -> Self {
195        self.credential_providers = Some(chain);
196        self
197    }
198
199    /// Set authority_host of this backend.
200    ///
201    /// - If authority_host is set, we will take user's input first.
202    /// - If not, we will try to load it from environment.
203    /// - default value: `https://login.microsoftonline.com`
204    pub fn authority_host(mut self, authority_host: &str) -> Self {
205        if !authority_host.is_empty() {
206            self.config.authority_host = Some(authority_host.to_string());
207        }
208
209        self
210    }
211
212    /// Create a new `AzdlsBuilder` instance from an [Azure Storage connection string][1].
213    ///
214    /// [1]: https://learn.microsoft.com/en-us/azure/storage/common/storage-configure-connection-string
215    ///
216    /// # Example
217    /// ```
218    /// use opendal_core::Builder;
219    /// use opendal_service_azdls::Azdls;
220    ///
221    /// let conn_str = "AccountName=example;DefaultEndpointsProtocol=https;EndpointSuffix=core.windows.net";
222    ///
223    /// let mut config = Azdls::from_connection_string(&conn_str)
224    ///     .unwrap()
225    ///     // Add additional configuration if needed
226    ///     .filesystem("myFilesystem")
227    ///     .client_id("myClientId")
228    ///     .client_secret("myClientSecret")
229    ///     .tenant_id("myTenantId")
230    ///     .build()
231    ///     .unwrap();
232    /// ```
233    pub fn from_connection_string(conn_str: &str) -> Result<Self> {
234        let config = azure_config_from_connection_string(conn_str, AzureStorageService::Adls)?;
235
236        Ok(AzdlsConfig::from(config).into_builder())
237    }
238
239    /// Enable or disable HNS (Hierarchical Namespace) for this backend.
240    pub fn enable_hns(mut self, enable: bool) -> Self {
241        self.config.enable_hns = enable;
242        self
243    }
244}
245
246impl Builder for AzdlsBuilder {
247    type Config = AzdlsConfig;
248
249    fn build(self) -> Result<impl Service> {
250        debug!("backend build started: {:?}", self);
251
252        let root = normalize_root(&self.config.root.unwrap_or_default());
253        debug!("backend use root {root}");
254
255        // Handle endpoint, region and container name.
256        let filesystem = match self.config.filesystem.is_empty() {
257            false => Ok(&self.config.filesystem),
258            true => Err(Error::new(ErrorKind::ConfigInvalid, "filesystem is empty")
259                .with_operation("Builder::build")
260                .with_context("service", AZDLS_SCHEME)),
261        }?;
262        debug!("backend use filesystem {}", filesystem);
263
264        let endpoint = match &self.config.endpoint {
265            Some(endpoint) => Ok(endpoint.clone().trim_end_matches('/').to_string()),
266            None => Err(Error::new(ErrorKind::ConfigInvalid, "endpoint is empty")
267                .with_operation("Builder::build")
268                .with_context("service", AZDLS_SCHEME)),
269        }?;
270        debug!("backend use endpoint {}", endpoint);
271
272        let account_name = self
273            .config
274            .account_name
275            .clone()
276            .or_else(|| azure_account_name_from_endpoint(endpoint.as_str()));
277
278        let mut envs = std::collections::HashMap::new();
279
280        if let Some(v) = &account_name {
281            envs.insert("AZBLOB_ACCOUNT_NAME".to_string(), v.clone());
282            envs.insert("AZURE_STORAGE_ACCOUNT_NAME".to_string(), v.clone());
283        }
284        if let Some(v) = &self.config.account_key {
285            envs.insert("AZBLOB_ACCOUNT_KEY".to_string(), v.clone());
286            envs.insert("AZURE_STORAGE_ACCOUNT_KEY".to_string(), v.clone());
287        }
288        if let Some(v) = &self.config.sas_token {
289            envs.insert("AZURE_STORAGE_SAS_TOKEN".to_string(), v.clone());
290        }
291        if let Some(v) = &self.config.client_id {
292            envs.insert("AZURE_CLIENT_ID".to_string(), v.clone());
293        }
294        if let Some(v) = &self.config.client_secret {
295            envs.insert("AZURE_CLIENT_SECRET".to_string(), v.clone());
296        }
297        if let Some(v) = &self.config.tenant_id {
298            envs.insert("AZURE_TENANT_ID".to_string(), v.clone());
299        }
300        if let Some(v) = &self.config.authority_host {
301            envs.insert("AZURE_AUTHORITY_HOST".to_string(), v.clone());
302        }
303
304        let os_env = OsEnv;
305        let ctx = Context::new()
306            .with_file_read(TokioFileRead)
307            .with_env(StaticEnv {
308                home_dir: os_env.home_dir(),
309                envs,
310            });
311
312        let mut credential_providers =
313            ProvideCredentialChain::new().push(DefaultCredentialProvider::new());
314
315        if let (Some(account_name), Some(account_key)) =
316            (account_name.as_deref(), self.config.account_key.as_deref())
317        {
318            credential_providers = credential_providers.push_front(
319                StaticCredentialProvider::new_shared_key(account_name, account_key),
320            );
321        }
322        if let Some(sas_token) = self.config.sas_token.as_deref() {
323            credential_providers =
324                credential_providers.push_front(StaticCredentialProvider::new_sas_token(sas_token));
325        }
326
327        if let Some(customized_credential_chain) = self.credential_providers {
328            credential_providers = customized_credential_chain;
329        }
330
331        let sign_ctx = ctx;
332        let signer = Signer::new(sign_ctx.clone(), credential_providers, RequestSigner::new());
333
334        let info = ServiceInfo::new(AZDLS_SCHEME, &root, filesystem);
335        let capability = Capability {
336            stat: true,
337            stat_with_if_match: true,
338            stat_with_if_none_match: true,
339            stat_with_if_modified_since: true,
340            stat_with_if_unmodified_since: true,
341
342            read: true,
343            read_with_if_match: true,
344            read_with_if_none_match: true,
345            read_with_if_modified_since: true,
346            read_with_if_unmodified_since: true,
347
348            write: true,
349            write_can_append: true,
350            write_can_multi: true,
351            write_with_if_none_match: true,
352            write_with_if_not_exists: true,
353            write_with_user_metadata: true,
354
355            create_dir: true,
356
357            delete: true,
358            delete_with_if_match: true,
359            delete_with_recursive: true,
360
361            rename: true,
362
363            list: true,
364
365            shared: true,
366
367            ..Default::default()
368        };
369
370        Ok(AzdlsBackend {
371            core: Arc::new(AzdlsCore {
372                info,
373                capability,
374                filesystem: self.config.filesystem.clone(),
375                root,
376                endpoint,
377                enable_hns: self.config.enable_hns,
378                signer,
379                sign_ctx,
380            }),
381        })
382    }
383}
384
385/// Backend for azblob services.
386#[derive(Debug, Clone)]
387pub struct AzdlsBackend {
388    pub(crate) core: Arc<AzdlsCore>,
389}
390
391impl Service for AzdlsBackend {
392    type Reader = oio::StreamReader<AzdlsReader>;
393    type Writer = AzdlsWriters;
394    type Lister = oio::PageLister<AzdlsLister>;
395    type Deleter = oio::OneShotDeleter<AzdlsDeleter>;
396    type Copier = ();
397    type Composer = ();
398
399    fn info(&self) -> ServiceInfo {
400        self.core.info.clone()
401    }
402
403    fn capability(&self) -> Capability {
404        self.core.capability
405    }
406
407    async fn create_dir(
408        &self,
409        ctx: &OperationContext,
410        path: &str,
411        _: OpCreateDir,
412    ) -> Result<RpCreateDir> {
413        let resp = self
414            .core
415            .azdls_create(ctx, path, DIRECTORY, &OpWrite::default())
416            .await?;
417
418        let status = resp.status();
419        match status {
420            StatusCode::CREATED | StatusCode::OK => Ok(RpCreateDir::default()),
421            _ => Err(parse_error(
422                ErrorContext::new(ServiceOperation("CreateDirectory")),
423                resp,
424            )),
425        }
426    }
427
428    async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
429        // Stat root always returns a DIR.
430        // TODO: include metadata for the root (#4746)
431        if path == "/" {
432            return Ok(RpStat::new(MetadataBuilder::dir().build()));
433        }
434
435        let metadata = self.core.azdls_stat_metadata(ctx, path, &args).await?;
436        Ok(RpStat::new(metadata))
437    }
438    fn read(&self, ctx: &OperationContext, path: &str, args: OpRead) -> Result<Self::Reader> {
439        let output: oio::StreamReader<AzdlsReader> = {
440            Ok(oio::StreamReader::new(AzdlsReader::new(
441                self.clone(),
442                ctx.clone(),
443                path,
444                args,
445            )))
446        }?;
447
448        Ok(output)
449    }
450
451    fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
452        let output: AzdlsWriters = {
453            if args.append() {
454                let w = AzdlsWriter::new(
455                    self.core.clone(),
456                    ctx.clone(),
457                    args.clone(),
458                    path.to_string(),
459                );
460                Ok(AzdlsWriters::Two(oio::AppendWriter::new(w)))
461            } else {
462                let w = AzdlsWriter::new(
463                    self.core.clone(),
464                    ctx.clone(),
465                    args.clone(),
466                    path.to_string(),
467                );
468                let w = oio::PositionWriter::new(
469                    ctx.executor().clone(),
470                    AzdlsLazyPositionWriter::new(w),
471                    args.concurrent(),
472                );
473                Ok(AzdlsWriters::One(w))
474            }
475        }?;
476
477        Ok(output)
478    }
479
480    fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter> {
481        let output: oio::OneShotDeleter<AzdlsDeleter> = {
482            Ok(oio::OneShotDeleter::new(AzdlsDeleter::new(
483                self.core.clone(),
484                ctx.clone(),
485            )))
486        }?;
487
488        Ok(output)
489    }
490
491    fn list(&self, ctx: &OperationContext, path: &str, args: OpList) -> Result<Self::Lister> {
492        let output: oio::PageLister<AzdlsLister> = {
493            let l = AzdlsLister::new(
494                self.core.clone(),
495                ctx.clone(),
496                path.to_string(),
497                args.limit(),
498            );
499
500            Ok(oio::PageLister::new(l))
501        }?;
502
503        Ok(output)
504    }
505
506    fn copy(
507        &self,
508        _ctx: &OperationContext,
509        _from: &str,
510        _to: &str,
511        _args: OpCopy,
512    ) -> Result<Self::Copier> {
513        Err(Error::new(
514            ErrorKind::Unsupported,
515            "operation is not supported",
516        ))
517    }
518
519    async fn rename(
520        &self,
521        ctx: &OperationContext,
522        from: &str,
523        to: &str,
524        _args: OpRename,
525    ) -> Result<RpRename> {
526        if let Some(resp) = self.core.azdls_ensure_parent_path(ctx, to).await? {
527            let status = resp.status();
528            match status {
529                StatusCode::CREATED | StatusCode::CONFLICT => {}
530                _ => {
531                    return Err(parse_error(
532                        ErrorContext::new(ServiceOperation("CreateDirectory")),
533                        resp,
534                    ));
535                }
536            }
537        }
538
539        let resp = self.core.azdls_rename(ctx, from, to).await?;
540
541        let status = resp.status();
542
543        match status {
544            StatusCode::CREATED => Ok(RpRename::default()),
545            _ => Err(parse_error(
546                ErrorContext::new(ServiceOperation("RenamePath")),
547                resp,
548            )),
549        }
550    }
551
552    async fn presign(
553        &self,
554        _ctx: &OperationContext,
555        _path: &str,
556        _args: OpPresign,
557    ) -> Result<RpPresign> {
558        Err(Error::new(
559            ErrorKind::Unsupported,
560            "operation is not supported",
561        ))
562    }
563}
opendal_service_azdls/backend.rs

opendal_service_azdls/
backend.rs
