add "create path" operation (files only for now) to Azure Storage data_lake (using Pipeline architecture) (#358)

* Ignore IDEA files.

* Very rough draft.

* Fix compiler error: align types of pipeline and pipeline_context.

* Fix compiler error: align types of pipeline and pipeline_context.

* Fix compiler error around request builder.

* Match create_collection implementation.

* Fix error conversion.

* Fix data lake example.

* Format code.

* Fix typo.

* Test path create. Failing...

* Add missing content length header.

* Clean up comment.

* Remove commented code. Add timestamp to file system name. Rename variables. Fix newlines. Don't panic when create path fails.

* WIP

* Fail e2e test if create path did not work, while still deleting file system.

* Use actual bearer token in auth header.

* Use actual bearer token.

* Code format.

* Remove unneeded bearer token reference from DataLakeClient.

* Print debug info.

* Send proper auth header.

* Code format.

* Code format.

* Adopt new way to do options.

* Remove TODO comment.

* No need to set body to empty byte array.

* Remove comment.

* Fix env var names.

* Improve create_path response headers.

* Ignore batch files.

* Add assertions to e2e test.

* Remove unused.

* Format code. Fix bug.

* Consolidate two examples into one. Align example with e2e test.

* Add step to e2e test and example that fails to create file that already exists.

* Add mock testing framework support code. But can't switch e2e test until all operations are using pipeline architecture.

* Introduce actual DataLakeContext.

Author: thovoll

.gitignore

@@ -16,3 +16,6 @@ sample/
 __azurite*.json
 __blobstorage__
 __queuestorage__
+.idea/
+*.iml
+*.bat

sdk/core/src/request_options/content_length.rs

@@ -0,0 +1,36 @@
+use crate::AddAsHeader;
+use http::request::Builder;
+
+#[derive(Debug, Clone, Copy)]
+pub struct ContentLength(i32);
+
+impl ContentLength {
+    /// Create a new `ContentLength`
+    pub fn new(count: i32) -> Self {
+        Self(count)
+    }
+}
+
+impl AddAsHeader for ContentLength {
+    fn add_as_header(&self, builder: Builder) -> Builder {
+        if self.0 <= 0 {
+            builder.header(http::header::CONTENT_LENGTH, -1)
+        } else {
+            builder.header(http::header::CONTENT_LENGTH, self.0)
+        }
+    }
+
+    fn add_as_header2(
+        &self,
+        request: &mut crate::Request,
+    ) -> Result<(), crate::errors::HTTPHeaderError> {
+        if self.0 >= 0 {
+            let (header_name, header_value) = (http::header::CONTENT_LENGTH, self.0);
+            request
+                .headers_mut()
+                .append(header_name, http::HeaderValue::from(header_value));
+        };
+
+        Ok(())
+    }
+}

sdk/core/src/request_options/mod.rs

@@ -4,6 +4,7 @@ mod client_request_id;
 mod content_disposition;
 mod content_encoding;
 mod content_language;
+mod content_length;
 mod content_range;
 mod content_type;
 mod continuation;
@@ -35,6 +36,7 @@ pub use client_request_id::ClientRequestId;
 pub use content_disposition::ContentDisposition;
 pub use content_encoding::ContentEncoding;
 pub use content_language::ContentLanguage;
+pub use content_length::ContentLength;
 pub use content_range::ContentRange;
 pub use content_type::ContentType;
 pub use continuation::Continuation;

sdk/storage/Cargo.toml

@@ -13,6 +13,7 @@ categories = ["api-bindings"]
 edition = "2018"
 
 [dependencies]
+async-trait = "0.1"
 azure_core = { path = "../core", version = "0.1.0" }
 ring = "0.16"
 base64 = "0.13"
@@ -41,6 +42,7 @@ reqwest = "0.11"
 [features]
 default = ["account", "blob", "queue", "table", "data_lake"]
 test_e2e = ["account", "blob", "queue", "table", "data_lake"]
+mock_transport_framework = [ "azure_core/mock_transport_framework"]
 test_integration = ["account", "blob", "queue", "table", "data_lake"]
 account = []
 azurite_workaround = []

sdk/storage/examples/data_lake_00.rs

@@ -1,68 +1,127 @@
+use azure_core::prelude::IfMatchCondition;
 use azure_core::prelude::*;
+use azure_identity::token_credentials::DefaultCredential;
+use azure_identity::token_credentials::TokenCredential;
 use azure_storage::core::prelude::*;
 use azure_storage::data_lake::prelude::*;
+use chrono::Utc;
 use futures::stream::StreamExt;
 use std::error::Error;
 use std::num::NonZeroU32;
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
-    // First we retrieve the account name and master key from environment variables.
-    let account = std::env::var("ADSL_STORAGE_ACCOUNT")
-        .expect("Set env variable ADSL_STORAGE_ACCOUNT first!");
-    let master_key = std::env::var("ADSL_STORAGE_MASTER_KEY")
-        .expect("Set env variable ADSL_STORAGE_MASTER_KEY first!");
+    let account = std::env::var("ADLSGEN2_STORAGE_ACCOUNT")
+        .expect("Set env variable ADLSGEN2_STORAGE_ACCOUNT first!");
+    let master_key = std::env::var("ADLSGEN2_STORAGE_MASTER_KEY")
+        .expect("Set env variable ADLSGEN2_STORAGE_MASTER_KEY first!");
 
-    let file_system_name = std::env::args()
-        .nth(1)
-        .expect("please specify the file system name as first parameter");
+    let now = Utc::now();
+    let file_system_name = format!("azurerustsdk-datalake-example-{}", now.timestamp());
 
     let http_client = new_http_client();
 
     let storage_account_client =
         StorageAccountClient::new_access_key(http_client.clone(), &account, &master_key);
 
+    let resource_id = "https://storage.azure.com/";
+    println!("getting bearer token for '{}'...", resource_id);
+    let bearer_token = DefaultCredential::default().get_token(resource_id).await?;
+    println!("token expires on {}", bearer_token.expires_on);
+    println!();
+
     let data_lake = storage_account_client
         .as_storage_client()
-        .as_data_lake_client(account)?;
+        .as_data_lake_client(account, bearer_token.token.secret().to_owned())?;
+
+    let file_system = data_lake.as_file_system_client(&file_system_name)?;
 
-    let file_system = data_lake.as_file_system_client(file_system_name)?;
+    let mut fs_properties = Properties::new();
+    fs_properties.insert("AddedVia", "Azure SDK for Rust");
 
-    // let's add some metadata. We call them "properties" to be consistent with the REST API definition from
-    // [https://docs.microsoft.com/rest/api/storageservices/datalakestoragegen2/filesystem/create#request-headers](https://docs.microsoft.com/rest/api/storageservices/datalakestoragegen2/filesystem/create#request-headers)
-    let mut properties = Properties::new();
-    properties.insert("AddedVia", "Azure SDK for Rust");
-    properties.insert("CreatedAt", chrono::Utc::now().to_string());
-    let response = file_system
+    println!("creating file system '{}'...", &file_system_name);
+    let create_fs_response = file_system
         .create()
-        .properties(&properties)
+        .properties(&fs_properties)
         .execute()
         .await?;
-    println!("repsonse == {:?}", response);
+    println!("create file system response == {:?}", create_fs_response);
+    println!();
 
+    println!("listing file systems...");
     let mut stream = Box::pin(
         data_lake
             .list()
             .max_results(NonZeroU32::new(3).unwrap())
             .stream(),
     );
-
-    while let Some(response) = stream.next().await {
-        println!("response == {:?}\n\n", response);
+    while let Some(list_fs_response) = stream.next().await {
+        println!("list file system response == {:?}", list_fs_response);
+        println!();
     }
 
-    properties.insert("ModifiedBy", "Iota");
-    let response = file_system
-        .set_properties(Some(&properties))
+    println!("getting file system properties...");
+    let get_fs_props_response = file_system.get_properties().execute().await?;
+    println!(
+        "get file system properties response == {:?}",
+        get_fs_props_response
+    );
+    println!();
+
+    let file_name = "example-file.txt";
+
+    println!("creating path '{}'...", file_name);
+    let create_path_response = file_system
+        .create_path(Context::default(), file_name, CreatePathOptions::default())
+        .await?;
+    println!("create path response == {:?}", create_path_response);
+    println!();
+
+    println!("creating path '{}' (overwrite)...", file_name);
+    let create_path_response = file_system
+        .create_path(Context::default(), file_name, CreatePathOptions::default())
+        .await?;
+    println!("create path response == {:?}", create_path_response);
+    println!();
+
+    println!("creating path '{}' (do not overwrite)...", file_name);
+    let do_not_overwrite =
+        CreatePathOptions::new().if_match_condition(IfMatchCondition::NotMatch("*"));
+    let create_path_result = file_system
+        .create_path(Context::default(), file_name, do_not_overwrite)
+        .await;
+    println!(
+        "create path result (should fail) == {:?}",
+        create_path_result
+    );
+    println!();
+
+    println!("setting file system properties...");
+    fs_properties.insert("ModifiedBy", "Iota");
+    let set_fs_props_response = file_system
+        .set_properties(Some(&fs_properties))
         .execute()
         .await?;
-    println!("response == {:?}\n\n", response);
+    println!(
+        "set file system properties response == {:?}",
+        set_fs_props_response
+    );
+    println!();
+
+    println!("getting file system properties...");
+    let get_fs_props_response = file_system.get_properties().execute().await?;
+    println!(
+        "get file system properties response == {:?}",
+        get_fs_props_response
+    );
+    println!();
 
-    let response = file_system.get_properties().execute().await?;
-    println!("response == {:?}\n\n", response);
+    println!("deleting file system...");
+    let delete_fs_response = file_system.delete().execute().await?;
+    println!("delete file system response == {:?}", delete_fs_response);
+    println!();
 
-    let response = file_system.delete().execute().await?;
-    println!("response == {:?}\n\n", response);
+    println!("data lake example done.");
 
     Ok(())
 }

sdk/storage/src/core/errors.rs

@@ -67,6 +67,14 @@ pub enum Error {
     CRC64Not8BytesLong(usize),
     #[error("At least one of these headers must be present: {0:?}")]
     HeadersNotFound(Vec<String>),
+    #[error("error writing the header value: {0}")]
+    InvalidHeaderValue(#[from] azure_core::HTTPHeaderError),
+}
+
+impl From<azure_core::HttpError> for Error {
+    fn from(error: azure_core::HttpError) -> Self {
+        Self::CoreError(azure_core::Error::HttpError(error))
+    }
 }
 
 #[non_exhaustive]

sdk/storage/src/data_lake/authorization_policy.rs

@@ -0,0 +1,38 @@
+use azure_core::{PipelineContext, Policy, PolicyResult, Request, Response};
+use http::header::AUTHORIZATION;
+use http::HeaderValue;
+use std::sync::Arc;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct AuthorizationPolicy {
+    bearer_token: String,
+}
+
+impl AuthorizationPolicy {
+    pub(crate) fn new(bearer_token: String) -> Self {
+        Self { bearer_token }
+    }
+}
+
+#[derive(Debug, Copy, Clone, PartialEq, Eq)]
+pub(crate) struct DataLakeContext {}
+
+#[async_trait::async_trait]
+impl Policy<DataLakeContext> for AuthorizationPolicy {
+    async fn send(
+        &self,
+        ctx: &mut PipelineContext<DataLakeContext>,
+        request: &mut Request,
+        next: &[Arc<dyn Policy<DataLakeContext>>],
+    ) -> PolicyResult<Response> {
+        let auth_header_value = format!("Bearer {}", &self.bearer_token);
+
+        request
+            .headers_mut()
+            .append(AUTHORIZATION, HeaderValue::from_str(&auth_header_value)?);
+
+        // now next[0] is safe (will not panic) because we checked
+        // at the beginning of the function.
+        next[0].send(ctx, request, &next[1..]).await
+    }
+}