Skip to content
This repository was archived by the owner on Sep 6, 2023. It is now read-only.

Commit 58b5ade

Browse files
author
Lilian ARAGO
committed
Multiple uploadModes
1 parent 38f4d1d commit 58b5ade

11 files changed

Lines changed: 176 additions & 143 deletions

‎TCC.Lib/OperationBlock.cs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -236,6 +236,7 @@ public class StepResult
236236
public string Errors { get; set; }
237237
public string Warning { get; set; }
238238
public string Infos { get; set; }
239+
public UploadMode? UploadMode { get; set; }
239240
public bool IsSuccess => !HasError && !HasWarning;
240241
public bool HasError => !string.IsNullOrWhiteSpace(Errors);
241242
public bool HasWarning => !string.IsNullOrWhiteSpace(Warning);

‎TCC.Lib/Options/CompressOption.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
using System.Collections.Generic;
2+
using System.Linq;
23
using TCC.Lib.Blocks;
34
using TCC.Lib.Database;
45

@@ -27,6 +28,7 @@ public class CompressOption : TccOption
2728
public string S3Host { get; set; }
2829
public string S3BucketName { get; set; }
2930
public string S3Region { get; set; }
31+
public IEnumerable<UploadMode> UploadModes { get; set; } = Enumerable.Empty<UploadMode>();
3032
public UploadMode? UploadMode { get; set; }
3133
}
3234
}

‎TCC.Lib/Storage/AzureRemoteStorage.cs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,9 @@ public async Task<UploadResponse> UploadAsync(string targetPath, Stream data, Ca
2424
ErrorMessage = response.ReasonPhrase,
2525
RemoteFilePath = targetPath
2626
};
27-
}
27+
}
28+
29+
public UploadMode GetMode() => UploadMode.AzureSdk;
30+
2831
}
2932
}

‎TCC.Lib/Storage/GoogleRemoteStorage.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,8 @@ public async Task<UploadResponse> UploadAsync(string targetPath, Stream data, Ca
3838
IsSuccess = true,
3939
RemoteFilePath = targetPath
4040
};
41-
}
41+
}
42+
43+
public UploadMode GetMode() => UploadMode.GoogleCloudStorage;
4244
}
4345
}

‎TCC.Lib/Storage/IRemoteStorage.cs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,12 +8,14 @@ namespace TCC.Lib.Storage
88
public interface IRemoteStorage
99
{
1010
Task<UploadResponse> UploadAsync(string targetPath, Stream data, CancellationToken token);
11-
11+
1212
public async Task<UploadResponse> UploadAsync(FileInfo file, DirectoryInfo rootFolder, CancellationToken token)
1313
{
1414
string targetPath = file.GetRelativeTargetPathTo(rootFolder);
1515
await using FileStream uploadFileStream = File.OpenRead(file.FullName);
1616
return await UploadAsync(targetPath, uploadFileStream, token);
17-
}
17+
}
18+
19+
public UploadMode GetMode();
1820
}
1921
}

‎TCC.Lib/Storage/NoneRemoteStorage.cs‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@ public class NoneRemoteStorage : IRemoteStorage
99
public Task<UploadResponse> UploadAsync(string targetPath, Stream data, CancellationToken token)
1010
{
1111
return Task.FromResult(new UploadResponse { IsSuccess = true, RemoteFilePath = targetPath });
12-
}
12+
}
13+
14+
public UploadMode GetMode() => UploadMode.None;
1315
}
1416
}

‎TCC.Lib/Storage/RemoteStorageFactory.cs‎

Lines changed: 67 additions & 53 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,11 @@
55
using Google.Cloud.Storage.V1;
66
using Microsoft.Extensions.Logging;
77
using System;
8+
using System.Collections;
9+
using System.Collections.Generic;
10+
using System.ComponentModel;
811
using System.IO;
12+
using System.Linq;
913
using System.Text;
1014
using System.Threading;
1115
using System.Threading.Tasks;
@@ -16,59 +20,69 @@ namespace TCC.Lib.Storage
1620
{
1721
public static class RemoteStorageFactory
1822
{
19-
public static async Task<IRemoteStorage> GetRemoteStorageAsync(this CompressOption option, ILogger logger, CancellationToken token)
20-
{
21-
switch (option.UploadMode)
22-
{
23-
case UploadMode.AzureSdk:
24-
{
25-
if (string.IsNullOrEmpty(option.AzBlobUrl)
26-
|| string.IsNullOrEmpty(option.AzBlobContainer)
27-
|| string.IsNullOrEmpty(option.AzBlobSaS))
28-
{
29-
logger.LogCritical("Configuration error for azure blob upload");
30-
return new NoneRemoteStorage();
31-
}
32-
var client = new BlobServiceClient(new Uri(option.AzBlobUrl + "/" + option.AzBlobContainer + "?" + option.AzBlobSaS));
33-
BlobContainerClient container = client.GetBlobContainerClient(option.AzBlobContainer);
34-
return new AzureRemoteStorage(container);
35-
}
36-
case UploadMode.GoogleCloudStorage:
37-
{
38-
if (string.IsNullOrEmpty(option.GoogleStorageCredential)
39-
|| string.IsNullOrEmpty(option.GoogleStorageBucketName))
40-
{
41-
logger.LogCritical("Configuration error for google storage upload");
42-
return new NoneRemoteStorage();
43-
}
44-
StorageClient storage = await GoogleAuthHelper.GetGoogleStorageClientAsync(option.GoogleStorageCredential, token);
45-
return new GoogleRemoteStorage(storage, option.GoogleStorageBucketName);
46-
}
47-
case UploadMode.S3:
48-
if (string.IsNullOrEmpty(option.S3AccessKeyId)
49-
|| string.IsNullOrEmpty(option.S3Host)
50-
|| string.IsNullOrEmpty(option.S3Region)
51-
|| string.IsNullOrEmpty(option.S3BucketName)
52-
|| string.IsNullOrEmpty(option.S3SecretAcessKey))
53-
{
54-
logger.LogCritical("Configuration error for S3 upload");
55-
return new NoneRemoteStorage();
56-
}
57-
58-
var credentials = new BasicAWSCredentials(option.S3AccessKeyId, option.S3SecretAcessKey);
59-
var s3Config = new AmazonS3Config()
60-
{
61-
AuthenticationRegion = option.S3Region,
62-
ServiceURL = option.S3Host,
63-
};
64-
65-
return new S3RemoteStorage(new AmazonS3Client(credentials, s3Config), option.S3BucketName);
66-
case UploadMode.None:
67-
case null:
68-
return new NoneRemoteStorage();
69-
default:
70-
throw new ArgumentOutOfRangeException();
71-
}
23+
public static async Task<IEnumerable<IRemoteStorage>> GetRemoteStoragesAsync(this CompressOption option, ILogger logger, CancellationToken token)
24+
{
25+
var remoteStorages = new List<IRemoteStorage>();
26+
27+
option.UploadModes = option.UploadModes.Append(option.UploadMode ?? UploadMode.None).Distinct();
28+
29+
foreach(var mode in option.UploadModes)
30+
{
31+
switch (mode)
32+
{
33+
case UploadMode.AzureSdk:
34+
{
35+
if (string.IsNullOrEmpty(option.AzBlobUrl)
36+
|| string.IsNullOrEmpty(option.AzBlobContainer)
37+
|| string.IsNullOrEmpty(option.AzBlobSaS))
38+
{
39+
logger.LogCritical("Configuration error for azure blob upload");
40+
continue;
41+
}
42+
var client = new BlobServiceClient(new Uri(option.AzBlobUrl + "/" + option.AzBlobContainer + "?" + option.AzBlobSaS));
43+
BlobContainerClient container = client.GetBlobContainerClient(option.AzBlobContainer);
44+
remoteStorages.Add(new AzureRemoteStorage(container));
45+
break;
46+
}
47+
case UploadMode.GoogleCloudStorage:
48+
{
49+
if (string.IsNullOrEmpty(option.GoogleStorageCredential)
50+
|| string.IsNullOrEmpty(option.GoogleStorageBucketName))
51+
{
52+
logger.LogCritical("Configuration error for google storage upload");
53+
continue;
54+
}
55+
StorageClient storage = await GoogleAuthHelper.GetGoogleStorageClientAsync(option.GoogleStorageCredential, token);
56+
remoteStorages.Add(new GoogleRemoteStorage(storage, option.GoogleStorageBucketName));
57+
break;
58+
}
59+
case UploadMode.S3:
60+
if (string.IsNullOrEmpty(option.S3AccessKeyId)
61+
|| string.IsNullOrEmpty(option.S3Host)
62+
|| string.IsNullOrEmpty(option.S3Region)
63+
|| string.IsNullOrEmpty(option.S3BucketName)
64+
|| string.IsNullOrEmpty(option.S3SecretAcessKey))
65+
{
66+
logger.LogCritical("Configuration error for S3 upload");
67+
68+
}
69+
70+
var credentials = new BasicAWSCredentials(option.S3AccessKeyId, option.S3SecretAcessKey);
71+
var s3Config = new AmazonS3Config()
72+
{
73+
AuthenticationRegion = option.S3Region,
74+
ServiceURL = option.S3Host,
75+
};
76+
77+
remoteStorages.Add(new S3RemoteStorage(new AmazonS3Client(credentials, s3Config), option.S3BucketName));
78+
break;
79+
case UploadMode.None:
80+
break;
81+
default:
82+
throw new ArgumentOutOfRangeException();
83+
}
84+
}
85+
return remoteStorages;
7286
}
7387
}
7488
}

‎TCC.Lib/Storage/S3RemoteStorage.cs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,5 +44,7 @@ await _s3Client.PutObjectAsync(new Amazon.S3.Model.PutObjectRequest() {
4444
RemoteFilePath = targetPath
4545
};
4646
}
47+
48+
public UploadMode GetMode() => UploadMode.S3;
4749
}
4850
}

‎TCC.Lib/TarCompressCrypt.cs‎

Lines changed: 81 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -56,7 +56,7 @@ public async Task<OperationSummary> CompressAsync(CompressOption option)
5656
_logger.LogInformation("Starting compression job");
5757
var po = ParallelizeOption(option);
5858

59-
IRemoteStorage uploader = await option.GetRemoteStorageAsync(_logger, _cancellationTokenSource.Token);
59+
IEnumerable<IRemoteStorage> uploaders = await option.GetRemoteStoragesAsync(_logger, _cancellationTokenSource.Token);
6060

6161
var operationBlocks = await buffer
6262
.AsAsyncStream(_cancellationTokenSource.Token)
@@ -81,90 +81,93 @@ public async Task<OperationSummary> CompressAsync(CompressOption option)
8181
return opb;
8282
}, po)
8383
// Upload loop
84-
.ParallelizeStreamAsync((block, token) => UploadBlockInternal(uploader, option, block, token), new ParallelizeOption { FailMode = Fail.Smart, MaxDegreeOfParallelism = option.AzThread ?? 1 })
84+
.ParallelizeStreamAsync((block, token) => UploadBlockInternal(uploaders, option, block, token), new ParallelizeOption { FailMode = Fail.Smart, MaxDegreeOfParallelism = option.AzThread ?? 1 })
8585
.AsReadOnlyCollectionAsync();
8686

8787
sw.Stop();
8888
return new OperationSummary(operationBlocks, option.Threads, sw, option.SourceDirOrFile);
8989
}
9090

91-
private async Task<OperationCompressionBlock> UploadBlockInternal(IRemoteStorage uploader, CompressOption option, OperationCompressionBlock block, CancellationToken token)
92-
{
93-
if (uploader is NoneRemoteStorage)
94-
{
95-
return block;
91+
private async Task<OperationCompressionBlock> UploadBlockInternal(IEnumerable<IRemoteStorage> uploaders, CompressOption option, OperationCompressionBlock block, CancellationToken token)
92+
{
93+
foreach(var uploader in uploaders)
94+
{
95+
if (uploader is NoneRemoteStorage)
96+
{
97+
continue;
98+
}
99+
100+
int count = Interlocked.Increment(ref _uploadCounter);
101+
string progress = $"{count}/{_totalCounter}";
102+
103+
var file = block.CompressionBlock.DestinationArchiveFileInfo;
104+
var name = file.Name;
105+
RetryContext ctx = null;
106+
while (true)
107+
{
108+
bool hasError;
109+
try
110+
{
111+
var sw = Stopwatch.StartNew();
112+
113+
var result = await uploader.UploadAsync(file, block.CompressionBlock.FolderProvider.RootFolder, token);
114+
hasError = !result.IsSuccess;
115+
116+
sw.Stop();
117+
double speed = file.Length / sw.Elapsed.TotalSeconds;
118+
119+
block.BlockResult.StepResults.Add(new StepResult
120+
{
121+
Type = StepType.Upload,
122+
UploadMode = uploader.GetMode(),
123+
Errors = result.IsSuccess ? null : result.ErrorMessage,
124+
Infos = result.IsSuccess ? result.ErrorMessage : null,
125+
Duration = sw.Elapsed,
126+
ArchiveFileSize = file.Length,
127+
});
128+
129+
130+
if (!hasError)
131+
{
132+
_logger.LogInformation($"{progress} Uploaded \"{file.Name}\" in {sw.Elapsed.HumanizedTimeSpan()} at {speed.HumanizedBandwidth()} ");
133+
}
134+
else
135+
{
136+
if (ctx == null && option.RetryPeriodInSeconds.HasValue)
137+
{
138+
ctx = new RetryContext(option.RetryPeriodInSeconds.Value);
139+
}
140+
_logger.LogError($"{progress} Uploaded {file.Name} with errors. {result.ErrorMessage}");
141+
}
142+
}
143+
catch (Exception e)
144+
{
145+
hasError = true;
146+
if (ctx == null && option.RetryPeriodInSeconds.HasValue)
147+
{
148+
ctx = new RetryContext(option.RetryPeriodInSeconds.Value);
149+
}
150+
_logger.LogCritical(e, $"{progress} Error uploading {name}");
151+
}
152+
153+
if (hasError)
154+
{
155+
if (ctx != null && await ctx.WaitForNextRetry())
156+
{
157+
_logger.LogWarning($"{progress} Retrying uploading {name}, attempt #{ctx.Retries}");
158+
}
159+
else
160+
{
161+
break;
162+
}
163+
}
164+
else
165+
{
166+
break;
167+
}
168+
}
96169
}
97-
98-
int count = Interlocked.Increment(ref _uploadCounter);
99-
string progress = $"{count}/{_totalCounter}";
100-
101-
var file = block.CompressionBlock.DestinationArchiveFileInfo;
102-
var name = file.Name;
103-
RetryContext ctx = null;
104-
while (true)
105-
{
106-
bool hasError;
107-
try
108-
{
109-
var sw = Stopwatch.StartNew();
110-
111-
var result = await uploader.UploadAsync(file, block.CompressionBlock.FolderProvider.RootFolder, token);
112-
hasError = !result.IsSuccess;
113-
114-
sw.Stop();
115-
double speed = file.Length / sw.Elapsed.TotalSeconds;
116-
117-
block.BlockResult.StepResults.Add(new StepResult
118-
{
119-
Type = StepType.Upload,
120-
Errors = result.IsSuccess ? null : result.ErrorMessage,
121-
Infos = result.IsSuccess ? result.ErrorMessage : null,
122-
Duration = sw.Elapsed,
123-
ArchiveFileSize = file.Length,
124-
});
125-
126-
127-
if (!hasError)
128-
{
129-
_logger.LogInformation($"{progress} Uploaded \"{file.Name}\" in {sw.Elapsed.HumanizedTimeSpan()} at {speed.HumanizedBandwidth()} ");
130-
}
131-
else
132-
{
133-
if (ctx == null && option.RetryPeriodInSeconds.HasValue)
134-
{
135-
ctx = new RetryContext(option.RetryPeriodInSeconds.Value);
136-
}
137-
_logger.LogError($"{progress} Uploaded {file.Name} with errors. {result.ErrorMessage}");
138-
}
139-
}
140-
catch (Exception e)
141-
{
142-
hasError = true;
143-
if (ctx == null && option.RetryPeriodInSeconds.HasValue)
144-
{
145-
ctx = new RetryContext(option.RetryPeriodInSeconds.Value);
146-
}
147-
_logger.LogCritical(e, $"{progress} Error uploading {name}");
148-
}
149-
150-
if (hasError)
151-
{
152-
if (ctx != null && await ctx.WaitForNextRetry())
153-
{
154-
_logger.LogWarning($"{progress} Retrying uploading {name}, attempt #{ctx.Retries}");
155-
}
156-
else
157-
{
158-
break;
159-
}
160-
}
161-
else
162-
{
163-
break;
164-
}
165-
}
166-
167-
return block;
170+
return block;
168171
}
169172

170173
private async Task CleanupOldFiles(OperationCompressionBlock opb)

0 commit comments

Comments
 (0)