Adding ability to provide content type and encoding for Azure blobs (#887)

Author: anpaz

src/Azure/Azure.Quantum.Client/Microsoft.Azure.Quantum.Client.csproj

@@ -20,10 +20,10 @@
   </PropertyGroup>  
 
   <ItemGroup>
-    <PackageReference Include="Azure.Core" Version="1.14.0" />
+    <PackageReference Include="Azure.Core" Version="1.19.0" />
     <PackageReference Include="Azure.Identity" Version="1.4.0" />
     <PackageReference Include="Azure.Quantum.Jobs" Version="1.0.0-beta.2" />
-    <PackageReference Include="Azure.Storage.Blobs" Version="12.2.0" />
+    <PackageReference Include="Azure.Storage.Blobs" Version="12.10.0" />
     <PackageReference Include="Newtonsoft.Json" Version="12.0.3" />
     <PackageReference Include="StyleCop.Analyzers" Version="1.1.118">
       <PrivateAssets>all</PrivateAssets>

src/Azure/Azure.Quantum.Client/Storage/IJobStorageHelper.cs

@@ -29,6 +29,22 @@ public interface IJobStorageHelper
             Stream input,
             CancellationToken cancellationToken = default);
 
+        /// <summary>
+        /// Uploads the job input.
+        /// </summary>
+        /// <param name="jobId">The job id.</param>
+        /// <param name="input">The input.</param>
+        /// <param name="contentType">The MIME type indicating the content of the payload.</param>
+        /// <param name="compress">A flag to indicate if the payload should be uploaded compressed to storage.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>Container uri + Input uri.</returns>
+        Task<(string containerUri, string inputUri)> UploadJobInputAsync(
+            string jobId,
+            Stream input,
+            string contentType,
+            bool compress,
+            CancellationToken cancellationToken = default);
+
         /// <summary>
         /// Uploads the job program output mapping.
         /// </summary>

src/Azure/Azure.Quantum.Client/Storage/IStorageHelper.cs

@@ -1,6 +1,8 @@
-// Copyright (c) Microsoft Corporation. All rights reserved.
+// Copyright (c) Microsoft Corporation.
 // Licensed under the MIT License.
 
+#nullable enable
+
 namespace Microsoft.Azure.Quantum.Storage
 {
     using System;
@@ -32,14 +34,32 @@ Task DownloadBlobAsync(
         /// <param name="containerClient">Container client.</param>
         /// <param name="blobName">Name of the BLOB.</param>
         /// <param name="input">The input.</param>
+        /// <param name="contentType">The MIME type indicating the content of the payload.</param>
+        /// <param name="contentEncoding">The blob encoding.</param>
         /// <param name="cancellationToken">The cancellation token.</param>
         /// <returns>async task.</returns>
         Task UploadBlobAsync(
             BlobContainerClient containerClient,
             string blobName,
             Stream input,
+            string? contentType,
+            string? contentEncoding,
             CancellationToken cancellationToken = default);
 
+        /// <summary>
+        /// Uploads the BLOB.
+        /// </summary>
+        /// <param name="containerClient">Container client.</param>
+        /// <param name="blobName">Name of the BLOB.</param>
+        /// <param name="input">The input.</param>
+        /// <param name="cancellationToken">The cancellation token.</param>
+        /// <returns>async task.</returns>
+        Task UploadBlobAsync(
+            BlobContainerClient containerClient,
+            string blobName,
+            Stream input,
+            CancellationToken cancellationToken = default) => this.UploadBlobAsync(containerClient, blobName, input, null, null, cancellationToken);
+
         /// <summary>
         /// Gets the BLOB sas URI.
         /// </summary>

src/Azure/Azure.Quantum.Client/Storage/JobStorageHelper.cs

@@ -37,28 +37,34 @@ public JobStorageHelper(string connectionString)
             }
         }
 
-        /// <summary>
-        /// Uploads the job input.
-        /// </summary>
-        /// <param name="jobId">The job id.</param>
-        /// <param name="input">The input.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>
-        /// Container uri + Input uri.
-        /// </returns>
+        /// <inheritdoc/>
         public override async Task<(string containerUri, string inputUri)> UploadJobInputAsync(
             string jobId,
             Stream input,
+            string contentType,
+            bool compress,
             CancellationToken cancellationToken = default)
         {
             string containerName = GetContainerName(jobId);
+            string encoding = null;
+            Stream data = input;
+
+            if (compress)
+            {
+                var compressedInput = new MemoryStream();
+                await Compression.Compress(input, compressedInput);
+                data = compressedInput;
+                encoding = "gzip";
+            }
 
             BlobContainerClient containerClient = await this.GetContainerClient(containerName);
 
             await this.StorageHelper.UploadBlobAsync(
                 containerClient,
                 Constants.Storage.InputBlobName,
-                input,
+                input: data,
+                contentType: contentType,
+                contentEncoding: encoding,
                 cancellationToken);
 
             string containerUri = this.StorageHelper.GetBlobContainerSasUri(

src/Azure/Azure.Quantum.Client/Storage/JobStorageHelperBase.cs

@@ -7,7 +7,9 @@ namespace Microsoft.Azure.Quantum.Storage
     using System.IO;
     using System.Threading;
     using System.Threading.Tasks;
+
     using global::Azure.Storage.Blobs;
+
     using Microsoft.Azure.Quantum.Utility;
 
     public abstract class JobStorageHelperBase : IJobStorageHelper
@@ -54,17 +56,20 @@ await this.StorageHelper.DownloadBlobAsync(
             return;
         }
 
-        /// <summary>
-        /// Uploads the job input.
-        /// </summary>
-        /// <param name="jobId">The job id.</param>
-        /// <param name="input">The input.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>Container uri + Input uri.</returns>
+        /// <inheritdoc/>
         public abstract Task<(string containerUri, string inputUri)> UploadJobInputAsync(
-            string jobId,
-            Stream input,
-            CancellationToken cancellationToken = default);
+                string jobId,
+                Stream input,
+                string contentType,
+                bool compress,
+                CancellationToken cancellationToken = default);
+
+        /// <inheritdoc/>
+        public Task<(string containerUri, string inputUri)> UploadJobInputAsync(
+                string jobId,
+                Stream input,
+                CancellationToken cancellationToken = default) =>
+            this.UploadJobInputAsync(jobId, input, null, false, cancellationToken);
 
         /// <summary>
         /// Uploads the job program output mapping.

src/Azure/Azure.Quantum.Client/Storage/LinkedStorageJobHelper.cs

@@ -23,28 +23,34 @@ public LinkedStorageJobHelper(IWorkspace workspace)
             this.workspace = workspace;
         }
 
-        /// <summary>
-        /// Uploads the job input.
-        /// </summary>
-        /// <param name="jobId">The job id.</param>
-        /// <param name="input">The input.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>
-        /// Container uri + Input uri without SAS.
-        /// </returns>
+        /// <inheritdoc/>
         public override async Task<(string containerUri, string inputUri)> UploadJobInputAsync(
             string jobId,
             Stream input,
+            string contentType,
+            bool compress,
             CancellationToken cancellationToken = default)
         {
             string containerName = GetContainerName(jobId);
+            string encoding = null;
+            Stream data = input;
 
             BlobContainerClient containerClient = await this.GetContainerClient(containerName);
 
+            if (compress)
+            {
+                var compressedInput = new MemoryStream();
+                await Compression.Compress(input, compressedInput);
+                data = compressedInput;
+                encoding = "gzip";
+            }
+
             await this.StorageHelper.UploadBlobAsync(
-                containerClient,
-                Constants.Storage.InputBlobName,
-                input,
+                containerClient: containerClient,
+                blobName: Constants.Storage.InputBlobName,
+                input: data,
+                contentType: contentType,
+                contentEncoding: encoding,
                 cancellationToken);
 
             Uri inputUri = containerClient
@@ -54,13 +60,7 @@ await this.StorageHelper.UploadBlobAsync(
             return (GetUriPath(containerClient.Uri), GetUriPath(inputUri));
         }
 
-        /// <summary>
-        /// Uploads the job program output mapping.
-        /// </summary>
-        /// <param name="jobId">The job id.</param>
-        /// <param name="mapping">The job program output mapping.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>Container uri + Mapping uri without SAS.</returns>
+        /// <inheritdoc/>
         public override async Task<(string containerUri, string mappingUri)> UploadJobMappingAsync(
             string jobId,
             Stream mapping,

src/Azure/Azure.Quantum.Client/Storage/StorageHelper.cs

@@ -5,6 +5,8 @@ namespace Microsoft.Azure.Quantum.Storage
 {
     using System;
     using System.IO;
+    using System.Net.Mime;
+    using System.Text;
     using System.Threading;
     using System.Threading.Tasks;
     using global::Azure.Storage.Blobs;
@@ -42,28 +44,34 @@ public async Task DownloadBlobAsync(
             return;
         }
 
-        /// <summary>
-        /// Uploads the BLOB.
-        /// </summary>
-        /// <param name="containerClient">Container client.</param>
-        /// <param name="blobName">Name of the BLOB.</param>
-        /// <param name="input">The input.</param>
-        /// <param name="cancellationToken">The cancellation token.</param>
-        /// <returns>Async task.</returns>
+        /// <inheritdoc/>
         public async Task UploadBlobAsync(
             BlobContainerClient containerClient,
             string blobName,
             Stream input,
+            string contentType,
+            string contentEncoding,
             CancellationToken cancellationToken = default)
         {
             try
             {
                 // Ensure container is created
                 await containerClient.CreateIfNotExistsAsync(PublicAccessType.None, cancellationToken: cancellationToken);
 
+                var headers = new BlobHttpHeaders
+                {
+                    ContentEncoding = contentEncoding,
+                    ContentType = contentType,
+                };
+
+                var options = new BlobUploadOptions
+                {
+                    HttpHeaders = headers,
+                };
+
                 // Upload blob
                 BlobClient blob = containerClient.GetBlobClient(blobName);
-                await blob.UploadAsync(input, overwrite: true, cancellationToken);
+                await blob.UploadAsync(input, options, cancellationToken);
             }
             catch (Exception ex)
             {

src/Azure/Azure.Quantum.Client/Utility/Compression.cs

@@ -0,0 +1,27 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+namespace Microsoft.Azure.Quantum.Utility
+{
+    using System.IO;
+    using System.IO.Compression;
+    using System.Threading.Tasks;
+
+    internal class Compression
+    {
+        public static async Task Compress(Stream data, Stream compressedData)
+        {
+            using (var auxStream = new MemoryStream())
+            using (var zipStream = new GZipStream(auxStream, CompressionMode.Compress))
+            {
+                await data.CopyToAsync(zipStream);
+                await zipStream.FlushAsync();
+                auxStream.Position = 0;
+                await auxStream.CopyToAsync(compressedData);
+                await compressedData.FlushAsync();
+            }
+
+            compressedData.Position = 0;
+        }
+    }
+}