S3 storage implementation

TCC.Lib/Helpers/StringExtensions.cs

@@ -1,7 +1,8 @@
 using System;
 using System.Globalization;
 using System.IO;
-
+using System.Linq;
+
 namespace TCC.Lib.Helpers
 {
     public static class StringExtensions
@@ -33,6 +34,21 @@ public static String HumanizeSize(this long size)
             int place = Convert.ToInt32(Math.Floor(Math.Log(bytes, 1024)));
             double num = Math.Round(bytes / Math.Pow(1024, place), 1);
             return (Math.Sign(size) * num).ToString(CultureInfo.InvariantCulture) + " " + suf[place];
+        }
+
+        public static long ParseSize(this string humanizedSize)
+        {
+            string[] suf = { "b", "ko", "mo", "go", "to", "po", "eo" };
+            var size = humanizedSize.Trim().ToLower(CultureInfo.InvariantCulture);
+            var number = string.Join("", size.Where(char.IsDigit));
+            var unit = size.Substring(size.Length - 2);
+            var pow = Array.IndexOf(suf, unit);
+
+            return pow switch
+            {
+                -1 => long.Parse(number, CultureInfo.InvariantCulture),
+                _ => long.Parse(number, CultureInfo.InvariantCulture) * (long)Math.Pow(1024L, pow)
+            };
         }
 
         public static string HumanizedTimeSpan(this TimeSpan t, int parts = 2)

TCC.Lib/OperationBlock.cs

@@ -236,6 +236,7 @@ public class StepResult
         public string Errors { get; set; }
         public string Warning { get; set; }
         public string Infos { get; set; }
+        public UploadMode? UploadMode { get; set; }
         public bool IsSuccess => !HasError && !HasWarning;
         public bool HasError => !string.IsNullOrWhiteSpace(Errors);
         public bool HasWarning => !string.IsNullOrWhiteSpace(Warning);

TCC.Lib/Options/CompressOption.cs

@@ -1,4 +1,5 @@
 using System.Collections.Generic;
+using System.Linq;
 using TCC.Lib.Blocks;
 using TCC.Lib.Database;
 
@@ -21,7 +22,15 @@ public class CompressOption : TccOption
         public string AzBlobSaS { get; set; }
         public int? AzThread { get; set; }
         public string GoogleStorageBucketName { get; set; }
-        public string GoogleStorageCredential { get; set; }
-        public UploadMode? UploadMode { get; set; }
+        public string GoogleStorageCredential { get; set; }
+        public string S3AccessKeyId { get; set; }
+        public string S3SecretAcessKey { get; set; }
+        public string S3Host { get; set; }
+        public string S3BucketName { get; set; }
+        public string S3Region { get; set; }
+        public string S3MultipartThreshold { get; set; }
+        public string S3MultipartSize { get; set; }
+        public IEnumerable<UploadMode> UploadModes { get; set; } = Enumerable.Empty<UploadMode>();
+        public UploadMode? UploadMode { get; set; }
     }
 }

TCC.Lib/Storage/AzureRemoteStorage.cs

@@ -24,6 +24,9 @@ public async Task<UploadResponse> UploadAsync(string targetPath, Stream data, Ca
                 ErrorMessage = response.ReasonPhrase,
                 RemoteFilePath = targetPath
             };
-        }
+        }
+
+        public UploadMode Mode => UploadMode.AzureSdk;
+
     }
 }

TCC.Lib/Storage/GoogleRemoteStorage.cs

@@ -38,6 +38,8 @@ public async Task<UploadResponse> UploadAsync(string targetPath, Stream data, Ca
                 IsSuccess = true,
                 RemoteFilePath = targetPath
             };
-        }
+        }
+
+        public UploadMode Mode => UploadMode.GoogleCloudStorage;
     }
 }

TCC.Lib/Storage/IRemoteStorage.cs

@@ -8,12 +8,14 @@ namespace TCC.Lib.Storage
     public interface IRemoteStorage
     {
         Task<UploadResponse> UploadAsync(string targetPath, Stream data, CancellationToken token);
-
+
         public async Task<UploadResponse> UploadAsync(FileInfo file, DirectoryInfo rootFolder, CancellationToken token)
         {
             string targetPath = file.GetRelativeTargetPathTo(rootFolder);
             await using FileStream uploadFileStream = File.OpenRead(file.FullName);
             return await UploadAsync(targetPath, uploadFileStream, token);
-        }
+        }
+
+        public UploadMode Mode { get; }
     }
 }

TCC.Lib/Storage/NoneRemoteStorage.cs

@@ -9,6 +9,8 @@ public class NoneRemoteStorage : IRemoteStorage
         public Task<UploadResponse> UploadAsync(string targetPath, Stream data, CancellationToken token)
         {
             return Task.FromResult(new UploadResponse { IsSuccess = true, RemoteFilePath = targetPath });
-        }
+        }
+
+        public UploadMode Mode => UploadMode.None;
     }
 }

TCC.Lib/Storage/RemoteStorageFactory.cs

@@ -1,10 +1,11 @@
-using Azure.Storage.Blobs;
-using Google.Apis.Auth.OAuth2;
+using Amazon.Runtime;
+using Amazon.S3;
+using Azure.Storage.Blobs;
 using Google.Cloud.Storage.V1;
 using Microsoft.Extensions.Logging;
 using System;
-using System.IO;
-using System.Text;
+using System.Collections.Generic;
+using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
 using TCC.Lib.Helpers;
@@ -13,41 +14,75 @@
 namespace TCC.Lib.Storage
 {
     public static class RemoteStorageFactory
-    {
-        public static async Task<IRemoteStorage> GetRemoteStorageAsync(this CompressOption option, ILogger logger, CancellationToken token)
-        {
-            switch (option.UploadMode)
-            {
-                case UploadMode.AzureSdk:
-                {
-                    if (string.IsNullOrEmpty(option.AzBlobUrl)
-                        || string.IsNullOrEmpty(option.AzBlobContainer)
-                        || string.IsNullOrEmpty(option.AzBlobSaS))
-                    {
-                        logger.LogCritical("Configuration error for azure blob upload");
-                        return new NoneRemoteStorage();
-                    }
-                    var client = new BlobServiceClient(new Uri(option.AzBlobUrl + "/" + option.AzBlobContainer + "?" + option.AzBlobSaS));
-                    BlobContainerClient container = client.GetBlobContainerClient(option.AzBlobContainer);
-                    return new AzureRemoteStorage(container);
-                }
-                case UploadMode.GoogleCloudStorage:
-                {
-                    if (string.IsNullOrEmpty(option.GoogleStorageCredential)
-                        || string.IsNullOrEmpty(option.GoogleStorageBucketName))
-                    {
-                        logger.LogCritical("Configuration error for google storage upload");
-                        return new NoneRemoteStorage();
-                    }
-                    StorageClient storage = await GoogleAuthHelper.GetGoogleStorageClientAsync(option.GoogleStorageCredential, token);
-                    return new GoogleRemoteStorage(storage, option.GoogleStorageBucketName);
-                }
-                case UploadMode.None:
-                case null:
-                    return new NoneRemoteStorage();
-                default:
-                    throw new ArgumentOutOfRangeException();
-            }
+    {
+        private static long ParseSizeInternal(string size) => string.IsNullOrWhiteSpace(size) ? 0 : size.ParseSize();
+        public static async Task<IEnumerable<IRemoteStorage>> GetRemoteStoragesAsync(this CompressOption option, ILogger logger, CancellationToken token)
+        {
+            var remoteStorages = new List<IRemoteStorage>();
+
+            option.UploadModes = option.UploadModes.Append(option.UploadMode ?? UploadMode.None).Distinct();
+
+            foreach(var mode in option.UploadModes)
+            {
+                switch (mode)
+                {
+                    case UploadMode.AzureSdk:
+                        {
+                            if (string.IsNullOrEmpty(option.AzBlobUrl)
+                                || string.IsNullOrEmpty(option.AzBlobContainer)
+                                || string.IsNullOrEmpty(option.AzBlobSaS))
+                            {
+                                logger.LogCritical("Configuration error for azure blob upload");
+                                continue;
+                            }
+                            var client = new BlobServiceClient(new Uri(option.AzBlobUrl + "/" + option.AzBlobContainer + "?" + option.AzBlobSaS));
+                            BlobContainerClient container = client.GetBlobContainerClient(option.AzBlobContainer);
+                            remoteStorages.Add(new AzureRemoteStorage(container));
+                            break;
+                        }
+                    case UploadMode.GoogleCloudStorage:
+                        {
+                            if (string.IsNullOrEmpty(option.GoogleStorageCredential)
+                                || string.IsNullOrEmpty(option.GoogleStorageBucketName))
+                            {
+                                logger.LogCritical("Configuration error for google storage upload");
+                                continue;
+                            }
+                            StorageClient storage = await GoogleAuthHelper.GetGoogleStorageClientAsync(option.GoogleStorageCredential, token);
+                            remoteStorages.Add(new GoogleRemoteStorage(storage, option.GoogleStorageBucketName));
+                            break;
+                        }
+                    case UploadMode.S3:
+                        if (string.IsNullOrEmpty(option.S3AccessKeyId)
+                            || string.IsNullOrEmpty(option.S3Host)
+                            || string.IsNullOrEmpty(option.S3Region)
+                            || string.IsNullOrEmpty(option.S3BucketName)
+                            || string.IsNullOrEmpty(option.S3SecretAcessKey))
+                        {
+                            logger.LogCritical("Configuration error for S3 upload");
+
+                        }
+
+                        var credentials = new BasicAWSCredentials(option.S3AccessKeyId, option.S3SecretAcessKey);
+                        var s3Config = new AmazonS3Config
+                        {
+                            AuthenticationRegion = option.S3Region,
+                            ServiceURL = option.S3Host,
+                        };
+
+                        remoteStorages.Add(new S3RemoteStorage(
+                            new AmazonS3Client(credentials, s3Config),
+                            option.S3BucketName,
+                            ParseSizeInternal(option.S3MultipartThreshold),
+                            (int) ParseSizeInternal(option.S3MultipartSize)));
+                        break;
+                    case UploadMode.None:
+                        break;
+                    default:
+                        throw new ArgumentOutOfRangeException();
+                }
+            }
+            return remoteStorages;
         }
     }
 }

TCC.Lib/Storage/S3RemoteStorage.cs

@@ -0,0 +1,115 @@
+using Amazon.S3;
+using Amazon.S3.Model;
+using System;
+using System.Collections.Generic;
+using System.IO;
+using System.Runtime.CompilerServices;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace TCC.Lib.Storage
+{
+    public class S3RemoteStorage : IRemoteStorage
+    {
+        internal string BucketName { get; }
+        private readonly AmazonS3Client _s3Client;
+        private readonly long _multipartTreshold;
+        private readonly int _partSize;
+
+        public S3RemoteStorage(AmazonS3Client s3Client, string bucketName, long multipartThreshold = 0, int partSize = 0)
+        {
+            BucketName = bucketName;
+            _s3Client = s3Client;
+            _multipartTreshold = multipartThreshold;
+            _partSize = partSize;
+        }
+
+        public async Task<UploadResponse> UploadAsync(string targetPath, Stream data, CancellationToken token)
+        {
+            try
+            {
+                if (_multipartTreshold != 0  && data.Length > _multipartTreshold)
+                {
+                    await UploadStreamToMultipartsAsync(targetPath, data, token);
+                } else
+                {
+                    await _s3Client.PutObjectAsync(new ()
+                    {
+                        BucketName = BucketName,
+                        Key = targetPath,
+                        InputStream = data,
+                    }, token);
+                }
+            }
+            catch (Exception e)
+            {
+                return new UploadResponse
+                {
+                    IsSuccess = false,
+                    RemoteFilePath = targetPath,
+                    ErrorMessage = e.Message
+                };
+            }
+            return new UploadResponse
+            {
+                IsSuccess = true,
+                RemoteFilePath = targetPath
+            };
+        }
+
+        private async Task UploadStreamToMultipartsAsync(string targetPath, Stream data, CancellationToken token)
+        {
+
+            var multipartUpload = await _s3Client.InitiateMultipartUploadAsync(new ()
+            {
+                BucketName = BucketName,
+                Key = targetPath,
+            }, token);
+            var partsETags = new List<PartETag>();
+            var partNumber = 1;
+
+            await foreach (var chunk in ChunkStreamAsync(data, token))
+            {
+                chunk.Position = 0;
+                var partUpload = await _s3Client.UploadPartAsync(new()
+                {
+                    BucketName = BucketName,
+                    PartNumber = partNumber++,
+                    Key = targetPath,
+                    UploadId = multipartUpload.UploadId,
+                    InputStream = chunk,
+                }, token);
+
+                partsETags.Add(new()
+                {
+                    ETag = partUpload.ETag,
+                    PartNumber = partUpload.PartNumber,
+                });
+            }
+            await _s3Client.CompleteMultipartUploadAsync(new()
+            {
+                BucketName = BucketName,
+                Key = targetPath,
+                UploadId = multipartUpload.UploadId,
+                PartETags = partsETags
+            }, token);
+        }
+        private async IAsyncEnumerable<Stream> ChunkStreamAsync(Stream data, [EnumeratorCancellation] CancellationToken token)
+        {
+            var buffer = new byte[_partSize];
+            int readBytes;
+
+            do
+            {
+                readBytes = await data.ReadAsync(buffer.AsMemory(0, _partSize), token);
+
+                var notTheSame = new MemoryStream(readBytes);
+                notTheSame.Write(buffer, 0, readBytes);
+
+                yield return notTheSame;
+            } while (readBytes >= _partSize);
+        }
+
+        public UploadMode Mode => UploadMode.S3;
+    }
+}

TCC.Lib/TCC.Lib.csproj

@@ -6,6 +6,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="AWSSDK.S3" Version="3.7.10.1" />
     <PackageReference Include="Azure.Storage.Blobs" Version="12.13.0" />
     <PackageReference Include="Google.Cloud.Storage.V1" Version="3.7.0" />
     <PackageReference Include="Microsoft.EntityFrameworkCore.Design" Version="6.0.7">