1145 lines
36 KiB
C#
1145 lines
36 KiB
C#
// Copyright Epic Games, Inc. All Rights Reserved.
|
|
|
|
using System;
|
|
using System.Buffers;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Net;
|
|
using System.Net.Http;
|
|
using System.Runtime.CompilerServices;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using System.Xml;
|
|
using Amazon;
|
|
using Amazon.Extensions.NETCore.Setup;
|
|
using Amazon.Runtime;
|
|
using Amazon.S3;
|
|
using Amazon.S3.Model;
|
|
using Amazon.S3.Transfer;
|
|
using Amazon.S3.Util;
|
|
using EpicGames.Horde.Storage;
|
|
using Jupiter.Common;
|
|
using Jupiter.Common.Implementation;
|
|
using Jupiter.Common.Utils;
|
|
using Jupiter.Implementation.Blob;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Logging;
|
|
using Microsoft.Extensions.Options;
|
|
using OpenTelemetry.Trace;
|
|
using Exception = System.Exception;
|
|
using KeyNotFoundException = System.Collections.Generic.KeyNotFoundException;
|
|
|
|
namespace Jupiter.Implementation
|
|
{
|
|
public class AmazonS3Store : IBlobStore, IMultipartBlobStore, IRegionalBlobStore
|
|
{
|
|
private readonly IAmazonS3 _amazonS3;
|
|
private readonly IBlobIndex _blobIndex;
|
|
private readonly INamespacePolicyResolver _namespacePolicyResolver;
|
|
private readonly ILogger<AmazonS3Store> _logger;
|
|
private readonly IServiceProvider _provider;
|
|
private readonly S3Settings _settings;
|
|
private readonly JupiterSettings _jupiterSettings;
|
|
private readonly ConcurrentDictionary<string, AmazonStorageBackend> _backends = new ConcurrentDictionary<string, AmazonStorageBackend>();
|
|
|
|
public AmazonS3Store(AmazonS3Factory amazonS3Factory, IOptionsMonitor<S3Settings> settings, IOptionsMonitor<JupiterSettings> jupiterSettings, IBlobIndex blobIndex, INamespacePolicyResolver namespacePolicyResolver, ILogger<AmazonS3Store> logger, IServiceProvider provider)
|
|
{
|
|
_amazonS3 = amazonS3Factory.Create();
|
|
_blobIndex = blobIndex;
|
|
_namespacePolicyResolver = namespacePolicyResolver;
|
|
_logger = logger;
|
|
_provider = provider;
|
|
_settings = settings.CurrentValue;
|
|
_jupiterSettings = jupiterSettings.CurrentValue;
|
|
}
|
|
|
|
private AmazonStorageBackend GetBackend(NamespaceId ns, string? site = null)
|
|
{
|
|
site ??= _jupiterSettings.CurrentSite;
|
|
|
|
string bucketName = GetBucketName(ns, site);
|
|
|
|
return _backends.GetOrAdd($"{ns}:{site}", x =>
|
|
{
|
|
string? awsRegion = null;
|
|
|
|
// Determine the region for the bucket if we're not dealing with the current site
|
|
if (site != _jupiterSettings.CurrentSite)
|
|
{
|
|
GetBucketLocationRequest bucketLocationRequest = new GetBucketLocationRequest() { BucketName = bucketName };
|
|
GetBucketLocationResponse bucketLocationResponse = _amazonS3.GetBucketLocationAsync(bucketLocationRequest).Result;
|
|
|
|
// us-east-1 and eu-west-1 don't contain their actual region identifiers for some reason
|
|
if (bucketLocationResponse.Location == S3Region.USEast1)
|
|
{
|
|
awsRegion = "us-east-1";
|
|
}
|
|
else if (bucketLocationResponse.Location == S3Region.EUWest1)
|
|
{
|
|
awsRegion = "eu-west-1";
|
|
}
|
|
else
|
|
{
|
|
awsRegion = bucketLocationResponse.Location.Value;
|
|
}
|
|
}
|
|
|
|
return ActivatorUtilities.CreateInstance<AmazonStorageBackend>(_provider, new AmazonStorageBackendConfiguration() { BucketName = bucketName, Region = awsRegion });
|
|
});
|
|
}
|
|
|
|
public async Task<Uri?> GetObjectByRedirectAsync(NamespaceId ns, BlobId identifier)
|
|
{
|
|
Uri? uri = await GetBackend(ns).GetReadRedirectAsync(identifier.AsS3Key());
|
|
|
|
return uri;
|
|
}
|
|
|
|
public async Task<BlobMetadata> GetObjectMetadataAsync(NamespaceId ns, BlobId identifier)
|
|
{
|
|
BlobMetadata? result = await GetBackend(ns).GetMetadataAsync(identifier.AsS3Key());
|
|
if (result == null)
|
|
{
|
|
throw new BlobNotFoundException(ns, identifier);
|
|
}
|
|
|
|
return result;
|
|
}
|
|
|
|
public async Task CopyBlobAsync(NamespaceId ns, NamespaceId targetNamespace, BlobId blobId)
|
|
{
|
|
await GetBackend(targetNamespace).CopyBlobAsync(GetBackend(ns), blobId.AsS3Key(), blobId.AsS3Key());
|
|
}
|
|
|
|
public async Task<Uri?> PutObjectWithRedirectAsync(NamespaceId ns, BlobId identifier)
|
|
{
|
|
Uri? uri = await GetBackend(ns).GetWriteRedirectAsync(identifier.AsS3Key());
|
|
|
|
return uri;
|
|
}
|
|
|
|
public async Task<BlobId> PutObjectAsync(NamespaceId ns, ReadOnlyMemory<byte> content, BlobId objectName)
|
|
{
|
|
await using MemoryStream stream = new MemoryStream(content.ToArray());
|
|
return await PutObjectAsync(ns, stream, objectName);
|
|
}
|
|
|
|
public async Task<BlobId> PutObjectAsync(NamespaceId ns, Stream stream, BlobId objectName)
|
|
{
|
|
await GetBackend(ns).WriteAsync(objectName.AsS3Key(), stream, CancellationToken.None);
|
|
return objectName;
|
|
}
|
|
|
|
public async Task<BlobId> PutObjectAsync(NamespaceId ns, byte[] content, BlobId objectName)
|
|
{
|
|
await using MemoryStream stream = new MemoryStream(content);
|
|
return await PutObjectAsync(ns, stream, objectName);
|
|
}
|
|
|
|
private string GetBucketName(NamespaceId ns, string site)
|
|
{
|
|
try
|
|
{
|
|
NamespacePolicy policy = _namespacePolicyResolver.GetPoliciesForNs(ns);
|
|
string storagePool = policy.StoragePool;
|
|
|
|
// Prefer explicitly defined bucket names for the current storage pool and region
|
|
if (_settings.SiteBucketsPerStoragePool.TryGetValue(storagePool, out Dictionary<string, string>? siteBuckets))
|
|
{
|
|
if (siteBuckets.TryGetValue(site, out string? bucketName))
|
|
{
|
|
return bucketName;
|
|
}
|
|
}
|
|
|
|
if (site != _jupiterSettings.CurrentSite)
|
|
{
|
|
throw new Exception("Unable to resolve bucket name for specific site. Please define regional bucket names in S3.SiteBucketsPerStoragePool");
|
|
}
|
|
|
|
// if the bucket to use for the storage pool has been overriden we use the override
|
|
if (_settings.StoragePoolBucketOverride.TryGetValue(storagePool, out string? containerOverride))
|
|
{
|
|
return containerOverride;
|
|
}
|
|
// by default we use the storage pool as a suffix to determine the bucket for that pool
|
|
string storagePoolSuffix = string.IsNullOrEmpty(storagePool) ? "" : $"-{storagePool}";
|
|
return $"{_settings.BucketName}{storagePoolSuffix}";
|
|
}
|
|
catch (KeyNotFoundException)
|
|
{
|
|
throw new NamespaceNotFoundException(ns);
|
|
}
|
|
}
|
|
|
|
public async Task<BlobContents> GetObjectAsync(NamespaceId ns, BlobId blob, LastAccessTrackingFlags flags = LastAccessTrackingFlags.DoTracking, bool supportsRedirectUri = false)
|
|
{
|
|
NamespacePolicy policies = _namespacePolicyResolver.GetPoliciesForNs(ns);
|
|
try
|
|
{
|
|
if (supportsRedirectUri && policies.AllowRedirectUris)
|
|
{
|
|
Uri? redirectUri = await GetBackend(ns).GetReadRedirectAsync(blob.AsS3Key());
|
|
if (redirectUri != null)
|
|
{
|
|
return new BlobContents(redirectUri);
|
|
}
|
|
}
|
|
|
|
BlobContents? contents = await GetBackend(ns).TryReadAsync(blob.AsS3Key(), flags);
|
|
if (contents == null)
|
|
{
|
|
throw new BlobNotFoundException(ns, blob);
|
|
}
|
|
return contents;
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// log information about the failed request except for 404 as its valid to not find objects in S3
|
|
if (e.StatusCode != HttpStatusCode.NotFound)
|
|
{
|
|
_logger.LogWarning("Exception raised from S3 {Exception}. {RequestId} {Id}", e, e.RequestId, e.AmazonId2);
|
|
}
|
|
|
|
// rethrow the exception, we just wanted to log more information about the failed request for further debugging
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public async Task<bool> ExistsAsync(NamespaceId ns, BlobId blobIdentifier, bool forceCheck)
|
|
{
|
|
NamespacePolicy policies = _namespacePolicyResolver.GetPoliciesForNs(ns);
|
|
if (_settings.UseBlobIndexForExistsCheck && policies.UseBlobIndexForSlowExists && !forceCheck)
|
|
{
|
|
return await _blobIndex.BlobExistsInRegionAsync(ns, blobIdentifier);
|
|
}
|
|
else
|
|
{
|
|
return await GetBackend(ns).ExistsAsync(blobIdentifier.AsS3Key(), CancellationToken.None);
|
|
}
|
|
}
|
|
|
|
public async Task DeleteNamespaceAsync(NamespaceId ns)
|
|
{
|
|
string bucketName = GetBucketName(ns, _jupiterSettings.CurrentSite);
|
|
try
|
|
{
|
|
await _amazonS3.DeleteBucketAsync(bucketName);
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// if the bucket does not exist we get a not found status code
|
|
if (e.StatusCode == HttpStatusCode.NotFound)
|
|
{
|
|
// deleting a none existent bucket is a success
|
|
return;
|
|
}
|
|
|
|
// something else happened, lets just process it as usual
|
|
}
|
|
}
|
|
|
|
public async IAsyncEnumerable<(BlobId, DateTime)> ListObjectsAsync(NamespaceId ns)
|
|
{
|
|
IStorageBackend backend = GetBackend(ns);
|
|
await foreach ((string path, DateTime time) in backend.ListAsync())
|
|
{
|
|
// ignore objects in the temp prefix
|
|
if (path.StartsWith("Temp", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
continue;
|
|
}
|
|
string identifierString = path.Substring(path.LastIndexOf("/", StringComparison.Ordinal) + 1);
|
|
yield return (new BlobId(identifierString), time);
|
|
}
|
|
}
|
|
|
|
public async Task DeleteObjectAsync(NamespaceId ns, BlobId blobIdentifier)
|
|
{
|
|
IStorageBackend backend = GetBackend(ns);
|
|
await backend.DeleteAsync(blobIdentifier.AsS3Key());
|
|
}
|
|
|
|
public async Task DeleteObjectAsync(IEnumerable<NamespaceId> namespaces, BlobId blob)
|
|
{
|
|
List<NamespaceId> namespaceIds = namespaces.ToList();
|
|
List<string> storagePools = namespaceIds.Select(ns => _namespacePolicyResolver.GetPoliciesForNs(ns).StoragePool).Distinct().ToList();
|
|
|
|
Dictionary<string, NamespaceId> storagePoolsToClean = storagePools.ToDictionary(storagePool => storagePool, storagePool => namespaceIds.FirstOrDefault(id => _namespacePolicyResolver.GetPoliciesForNs(id).StoragePool == storagePool));
|
|
|
|
foreach ((string _, NamespaceId ns) in storagePoolsToClean)
|
|
{
|
|
await GetBackend(ns).DeleteAsync(blob.AsS3Key(), CancellationToken.None);
|
|
}
|
|
}
|
|
|
|
#region IMultipartBlobStore
|
|
|
|
|
|
private const string TempMultipartPrefix = "Temp/Multipart";
|
|
public Task<string> StartMultipartUploadAsync(NamespaceId ns, string blobName)
|
|
{
|
|
return GetBackend(ns).StartMultipartUploadAsync($"{TempMultipartPrefix}/{blobName}");
|
|
}
|
|
|
|
public Task CompleteMultipartUploadAsync(NamespaceId ns, string blobName, string uploadId, List<string> partIds)
|
|
{
|
|
return GetBackend(ns).CompleteMultipartUploadAsync($"{TempMultipartPrefix}/{blobName}", uploadId, partIds);
|
|
}
|
|
|
|
public Task<Uri?> GetWriteRedirectForPartAsync(NamespaceId ns, string blobName, string uploadId, string partIdentifier)
|
|
{
|
|
return Task.FromResult(GetBackend(ns).GetWriteRedirectForPart($"{TempMultipartPrefix}/{blobName}", uploadId, partIdentifier));
|
|
}
|
|
|
|
public async Task<BlobContents?> GetMultipartObjectByNameAsync(NamespaceId ns, string blobName)
|
|
{
|
|
return await GetBackend(ns).TryReadAsync($"{TempMultipartPrefix}/{blobName}", LastAccessTrackingFlags.SkipTracking);
|
|
}
|
|
|
|
public async Task RenameMultipartBlobAsync(NamespaceId ns, string blobName, BlobId blobId)
|
|
{
|
|
await GetBackend(ns).RenameMultipartBlobAsync($"{TempMultipartPrefix}/{blobName}", blobId.AsS3Key());
|
|
}
|
|
public Task PutMultipartPartAsync(NamespaceId ns, string blobName, string uploadId, string partIdentifier, byte[] blob)
|
|
{
|
|
return GetBackend(ns).PutMultipartPartAsync($"{TempMultipartPrefix}/{blobName}", uploadId, partIdentifier, blob);
|
|
}
|
|
|
|
public List<MultipartByteRange> GetMultipartRanges(NamespaceId ns, string uploadId, ulong blobLength)
|
|
{
|
|
ulong countOfChunks = (blobLength / IdealS3ChunkSize) + 1;
|
|
|
|
if (Math.Min(countOfChunks, MaxS3Chunks) == MaxS3Chunks)
|
|
{
|
|
throw new Exception("Multipart blob would use more then max chunks in S3, this is not supported");
|
|
}
|
|
|
|
List<MultipartByteRange> parts = new();
|
|
ulong firstByte = 0;
|
|
ulong lastByte = IdealS3ChunkSize; // last byte is inclusive as per http range requests
|
|
|
|
for (int i = 1; i < (int)countOfChunks+1; i++)
|
|
{
|
|
// s3 parts start at 1 and max is 10_000 (inclusive)
|
|
// s3 part ids are simply incrementing numbers
|
|
parts.Add(new MultipartByteRange() {FirstByte = firstByte, LastByte = lastByte, PartId=i.ToString()});
|
|
|
|
firstByte += IdealS3ChunkSize;
|
|
lastByte = Math.Min(lastByte + IdealS3ChunkSize, blobLength);
|
|
}
|
|
|
|
return parts;
|
|
}
|
|
|
|
public MultipartLimits GetMultipartLimits(NamespaceId ns)
|
|
{
|
|
return new MultipartLimits { IdealChunkSize = IdealS3ChunkSize, MaxCountOfChunks = (int)MaxS3Chunks, MinChunkSize = 8 * 1024 * 1024 };
|
|
}
|
|
|
|
private const ulong MaxS3Chunks = 10_000;
|
|
private const int IdealS3ChunkSize = 32 * 1024 * 1024; // 32 MB parts - this means the largest file we can upload to S3 is 312GB
|
|
|
|
#endregion
|
|
|
|
#region IRegionalBlobStore
|
|
|
|
public async Task<BlobId> PutObjectAsync(NamespaceId ns, ReadOnlyMemory<byte> content, BlobId objectName, string region)
|
|
{
|
|
await using MemoryStream stream = new MemoryStream(content.ToArray());
|
|
return await PutObjectAsync(ns, stream, objectName, region);
|
|
}
|
|
|
|
public async Task<BlobId> PutObjectAsync(NamespaceId ns, Stream stream, BlobId objectName, string region)
|
|
{
|
|
await GetBackend(ns, region).WriteAsync(objectName.AsS3Key(), stream, CancellationToken.None);
|
|
return objectName;
|
|
}
|
|
|
|
public async Task<BlobId> PutObjectAsync(NamespaceId ns, byte[] content, BlobId objectName, string region)
|
|
{
|
|
await using MemoryStream stream = new MemoryStream(content);
|
|
return await PutObjectAsync(ns, stream, objectName, region);
|
|
}
|
|
|
|
public async Task DeleteObjectAsync(NamespaceId ns, BlobId blobId, string region)
|
|
{
|
|
IStorageBackend backend = GetBackend(ns, region);
|
|
await backend.DeleteAsync(blobId.AsS3Key());
|
|
}
|
|
|
|
public async Task ImportBlobAsync(NamespaceId ns, string sourceRegion, BlobId blobId)
|
|
{
|
|
AmazonStorageBackend source = GetBackend(ns, sourceRegion);
|
|
|
|
AmazonStorageBackend destination = GetBackend(ns);
|
|
await destination.EnsureBucketExistsAsync();
|
|
|
|
await destination.CopyBlobAsync(source, blobId.AsS3Key(), blobId.AsS3Key());
|
|
}
|
|
|
|
#endregion
|
|
}
|
|
|
|
public readonly struct AmazonStorageBackendConfiguration
|
|
{
|
|
public string BucketName { get; init; }
|
|
public string? Region { get; init; }
|
|
}
|
|
|
|
public class AmazonStorageBackend : IStorageBackend
|
|
{
|
|
private readonly IAmazonS3 _amazonS3;
|
|
private readonly string _bucketName;
|
|
private readonly IOptionsMonitor<S3Settings> _settings;
|
|
private readonly Tracer _tracer;
|
|
private readonly ILogger<AmazonStorageBackend> _logger;
|
|
private readonly BufferedPayloadFactory _payloadFactory;
|
|
private bool _bucketExistenceChecked;
|
|
private bool _bucketAccessPolicyApplied;
|
|
|
|
public AmazonStorageBackend(AmazonStorageBackendConfiguration configuration, AmazonS3Factory amazonS3Factory, IOptionsMonitor<S3Settings> settings, Tracer tracer, ILogger<AmazonStorageBackend> logger, BufferedPayloadFactory payloadFactory)
|
|
{
|
|
_amazonS3 = amazonS3Factory.Create(configuration.Region);
|
|
_bucketName = configuration.BucketName;
|
|
_settings = settings;
|
|
_tracer = tracer;
|
|
_logger = logger;
|
|
_payloadFactory = payloadFactory;
|
|
}
|
|
|
|
public async Task EnsureBucketExistsAsync(CancellationToken cancellationToken = default)
|
|
{
|
|
if (_settings.CurrentValue.CreateBucketIfMissing)
|
|
{
|
|
if (!_bucketExistenceChecked)
|
|
{
|
|
bool bucketExist = await AmazonS3Util.DoesS3BucketExistV2Async(_amazonS3, _bucketName);
|
|
if (!bucketExist)
|
|
{
|
|
PutBucketRequest putBucketRequest = new PutBucketRequest
|
|
{
|
|
BucketName = _bucketName,
|
|
UseClientRegion = true
|
|
};
|
|
|
|
try
|
|
{
|
|
await _amazonS3.PutBucketAsync(putBucketRequest, cancellationToken);
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
if (e.StatusCode == HttpStatusCode.Conflict)
|
|
{
|
|
// bucket already exists so no need to try and create it, most likely a race condition with another write operation that already tried to create the bucket at the same time in this or another instance
|
|
// as this will end up with the bucket existing we can just ignore the error
|
|
}
|
|
else
|
|
{
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
_bucketExistenceChecked = true;
|
|
}
|
|
}
|
|
|
|
if (_settings.CurrentValue.SetBucketPolicies && !_bucketAccessPolicyApplied)
|
|
{
|
|
// block all public access to the bucket
|
|
try
|
|
{
|
|
await _amazonS3.PutPublicAccessBlockAsync(new PutPublicAccessBlockRequest
|
|
{
|
|
BucketName = _bucketName,
|
|
PublicAccessBlockConfiguration = new PublicAccessBlockConfiguration()
|
|
{
|
|
RestrictPublicBuckets = true,
|
|
BlockPublicAcls = true,
|
|
BlockPublicPolicy = true,
|
|
IgnorePublicAcls = true,
|
|
}
|
|
}, cancellationToken);
|
|
|
|
_bucketAccessPolicyApplied = true;
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// if a conflicting operation is being applied to the public access block we just ignore it, as it will get reset the next time we run
|
|
if (e.StatusCode != HttpStatusCode.Conflict)
|
|
{
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
public async Task WriteAsync(string path, Stream stream, CancellationToken cancellationToken)
|
|
{
|
|
await EnsureBucketExistsAsync(cancellationToken);
|
|
|
|
if (_settings.CurrentValue.UseMultiPartUpload)
|
|
{
|
|
await WriteMultipartAsync(path, stream, cancellationToken);
|
|
}
|
|
else
|
|
{
|
|
PutObjectRequest request = new PutObjectRequest
|
|
{
|
|
BucketName = _bucketName,
|
|
Key = path,
|
|
InputStream = stream,
|
|
UseChunkEncoding = _settings.CurrentValue.UseChunkEncoding,
|
|
DisablePayloadSigning = _settings.CurrentValue.DisablePayloadChecksums,
|
|
DisableDefaultChecksumValidation = _settings.CurrentValue.DisablePayloadChecksums,
|
|
};
|
|
|
|
try
|
|
{
|
|
await _amazonS3.PutObjectAsync(request, cancellationToken);
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// if the same object is added twice S3 will raise a error, as we are content addressed we can just accept whichever of the objects so we can ignore that error
|
|
if (e.StatusCode == HttpStatusCode.Conflict)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (e.StatusCode == HttpStatusCode.TooManyRequests)
|
|
{
|
|
throw new ResourceHasToManyRequestsException(e);
|
|
}
|
|
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task WriteMultipartAsync(string path, Stream stream, CancellationToken cancellationToken)
|
|
{
|
|
FilesystemBufferedPayload? payload = null;
|
|
try
|
|
{
|
|
await using MemoryStream localStream = new MemoryStream();
|
|
string? filePath = null;
|
|
if (stream is FileStream fileStream)
|
|
{
|
|
filePath = fileStream.Name;
|
|
}
|
|
// use multipart transfers when buffer is larger than 16 MB, which is also the S3 default
|
|
else if (stream.Length > 16 * (long)Math.Pow(2, 20))
|
|
{
|
|
// will be chunked by TransferUtility
|
|
using FilesystemBufferedPayloadWriter writer = _payloadFactory.CreateFilesystemBufferedPayloadWriter("s3-upload");
|
|
{
|
|
await using Stream writableStream = writer.GetWritableStream();
|
|
await stream.CopyToAsync(writableStream, cancellationToken);
|
|
}
|
|
payload = writer.Done();
|
|
|
|
filePath = payload.TempFile.FullName;
|
|
}
|
|
else
|
|
{
|
|
// copy the stream as TransferUtility will dispose this stream if any issue happens making us unable to retry
|
|
await stream.CopyToAsync(localStream, cancellationToken);
|
|
}
|
|
|
|
const int MaxRetryAttempts = 3;
|
|
for (int i = 0; i < MaxRetryAttempts; i++)
|
|
{
|
|
using TransferUtility utility = new TransferUtility(_amazonS3);
|
|
try
|
|
{
|
|
if (filePath != null)
|
|
{
|
|
TransferUtilityUploadRequest uploadRequest = new TransferUtilityUploadRequest()
|
|
{
|
|
BucketName = _bucketName,
|
|
Key = path,
|
|
FilePath = filePath,
|
|
DisablePayloadSigning = _settings.CurrentValue.DisablePayloadChecksums,
|
|
DisableDefaultChecksumValidation = _settings.CurrentValue.DisablePayloadChecksums,
|
|
};
|
|
await utility.UploadAsync(uploadRequest, cancellationToken);
|
|
}
|
|
else
|
|
{
|
|
// reset the stream position so that we upload the entire stream each attempt
|
|
localStream.Position = 0;
|
|
|
|
TransferUtilityUploadRequest uploadRequest = new TransferUtilityUploadRequest()
|
|
{
|
|
BucketName = _bucketName,
|
|
Key = path,
|
|
InputStream = localStream,
|
|
DisablePayloadSigning = _settings.CurrentValue.DisablePayloadChecksums,
|
|
DisableDefaultChecksumValidation = _settings.CurrentValue.DisablePayloadChecksums,
|
|
};
|
|
await utility.UploadAsync(uploadRequest, cancellationToken);
|
|
}
|
|
|
|
break;
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// if the same object is added twice S3 will raise a error, as we are content addressed we can just accept whichever of the objects so we can ignore that error
|
|
if (e.StatusCode == HttpStatusCode.Conflict)
|
|
{
|
|
return;
|
|
}
|
|
|
|
if (e.StatusCode == HttpStatusCode.TooManyRequests)
|
|
{
|
|
throw new ResourceHasToManyRequestsException(e);
|
|
}
|
|
|
|
if (e.StatusCode == HttpStatusCode.ServiceUnavailable && (e.ErrorCode == "SlowDown" || e.ErrorCode == "503 SlowDown"))
|
|
{
|
|
throw new ResourceHasToManyRequestsException(e);
|
|
}
|
|
|
|
throw;
|
|
}
|
|
catch (HttpIOException e)
|
|
{
|
|
if (i < MaxRetryAttempts-1)
|
|
{
|
|
_logger.LogWarning("Http IO exception when writing multipart blob to S3, retrying");
|
|
// retry
|
|
continue;
|
|
}
|
|
|
|
// to many retries, give up
|
|
_logger.LogError(e, "Http IO exception when writing multipart blob to S3");
|
|
throw;
|
|
}
|
|
catch (IOException e)
|
|
{
|
|
if (i < MaxRetryAttempts-1)
|
|
{
|
|
_logger.LogWarning("Http IO exception when writing multipart blob to S3, retrying");
|
|
// retry
|
|
continue;
|
|
}
|
|
|
|
// to many retries, give up
|
|
_logger.LogError(e, "IO exception when writing multipart blob to S3");
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
payload?.Dispose();
|
|
}
|
|
}
|
|
|
|
private async Task<BlobContents?> GetMultipartAsync(string path, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
GetObjectRequest request = new GetObjectRequest { BucketName = _bucketName, Key = path, PartNumber = 1};
|
|
GetObjectResponse response = await _amazonS3.GetObjectAsync(request, cancellationToken);
|
|
// parts are only set of blobs that was uploaded using multipart, otherwise part 1 is the whole file
|
|
if (response.PartsCount > 1)
|
|
{
|
|
// object had parts, we download them in parallel
|
|
// -1 because parts start at 1
|
|
int partsCount = response.PartsCount.Value;
|
|
|
|
byte[][] parts = new byte[partsCount][];
|
|
long[] lengths = new long[partsCount];
|
|
|
|
try
|
|
{
|
|
{
|
|
parts[0] = ArrayPool<byte>.Shared.Rent((int)response.ContentLength);
|
|
lengths[0] = response.ContentLength;
|
|
await using MemoryStream ms = new MemoryStream(parts[0]);
|
|
await response.ResponseStream.CopyToAsync(ms, cancellationToken);
|
|
}
|
|
|
|
// multipart object, download it in parts, we already have the first part
|
|
await Parallel.ForEachAsync(Enumerable.Range(2, partsCount-1), cancellationToken, async (part, token) =>
|
|
{
|
|
GetObjectRequest partRequest = new GetObjectRequest { BucketName = _bucketName, Key = path, PartNumber = part };
|
|
GetObjectResponse partResponse = await _amazonS3.GetObjectAsync(partRequest, cancellationToken);
|
|
|
|
// the first part was retrieved in the initial request
|
|
parts[part-1] = ArrayPool<byte>.Shared.Rent((int)partResponse.ContentLength);
|
|
lengths[part-1] = partResponse.ContentLength;
|
|
await using MemoryStream ms = new MemoryStream(parts[part-1]);
|
|
await partResponse.ResponseStream.CopyToAsync(ms, cancellationToken);
|
|
});
|
|
|
|
long totalSize = lengths.Sum();
|
|
if (totalSize < _settings.CurrentValue.MultiPartMaxMemoryBufferSize)
|
|
{
|
|
Stream s = new MemoryStream(new byte[(int)totalSize]);
|
|
for (int i = 0; i < parts.Length; i++)
|
|
{
|
|
byte[] part = parts[i];
|
|
await s.WriteAsync(part, 0, (int)lengths[i], cancellationToken);
|
|
}
|
|
|
|
s.Position = 0;
|
|
return new BlobContents(s, totalSize);
|
|
}
|
|
else
|
|
{
|
|
// large payload - we have to buffer to file
|
|
using FilesystemBufferedPayloadWriter writer = _payloadFactory.CreateFilesystemBufferedPayloadWriter("s3-download");
|
|
|
|
{
|
|
await using Stream s = writer.GetWritableStream();
|
|
for (int i = 0; i < parts.Length; i++)
|
|
{
|
|
byte[] part = parts[i];
|
|
await s.WriteAsync(part, 0, (int)lengths[i], cancellationToken);
|
|
}
|
|
}
|
|
|
|
return new BlobContents(writer.Done());
|
|
}
|
|
}
|
|
finally
|
|
{
|
|
for (int i = 0; i < partsCount; i++)
|
|
{
|
|
// ReSharper disable once ConditionIsAlwaysTrueOrFalseAccordingToNullableAPIContract
|
|
// This should never be null but during exceptions we have seen it be null so to avoid throwing more exceptions that hide the original exceptions we just null check here
|
|
if (parts[i] != null)
|
|
{
|
|
ArrayPool<byte>.Shared.Return(parts[i]);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
else
|
|
{
|
|
return new BlobContents(response.ResponseStream, response.ContentLength);
|
|
}
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
if (e.ErrorCode == "NoSuchKey")
|
|
{
|
|
return null;
|
|
}
|
|
|
|
if (e.ErrorCode == "NoSuchBucket")
|
|
{
|
|
return null;
|
|
}
|
|
|
|
if (e.StatusCode == HttpStatusCode.TooManyRequests)
|
|
{
|
|
throw new ResourceHasToManyRequestsException(e);
|
|
}
|
|
|
|
throw;
|
|
}
|
|
}
|
|
|
|
public async Task<BlobContents?> TryReadAsync(string path, LastAccessTrackingFlags flags = LastAccessTrackingFlags.DoTracking, CancellationToken cancellationToken = default)
|
|
{
|
|
if (_settings.CurrentValue.UseMultiPartDownload)
|
|
{
|
|
return await GetMultipartAsync(path, cancellationToken);
|
|
}
|
|
else
|
|
{
|
|
GetObjectResponse response;
|
|
try
|
|
{
|
|
response = await _amazonS3.GetObjectAsync(_bucketName, path, cancellationToken);
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
if (e.ErrorCode == "NoSuchKey")
|
|
{
|
|
return null;
|
|
}
|
|
|
|
if (e.ErrorCode == "NoSuchBucket")
|
|
{
|
|
return null;
|
|
}
|
|
throw;
|
|
}
|
|
return new BlobContents(response.ResponseStream, response.ContentLength);
|
|
}
|
|
}
|
|
|
|
public async Task<bool> ExistsAsync(string path, CancellationToken cancellationToken)
|
|
{
|
|
try
|
|
{
|
|
await _amazonS3.GetObjectMetadataAsync(_bucketName, path, cancellationToken);
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// if the object does not exist we get a not found status code
|
|
if (e.StatusCode == HttpStatusCode.NotFound)
|
|
{
|
|
return false;
|
|
}
|
|
|
|
// something else happened, lets just process it as usual
|
|
}
|
|
|
|
return true;
|
|
}
|
|
|
|
public async IAsyncEnumerable<(string, DateTime)> ListAsync([EnumeratorCancellation] CancellationToken cancellationToken)
|
|
{
|
|
if (_settings.CurrentValue.PerPrefixListing)
|
|
{
|
|
List<string> hashPrefixes = new List<string>(65536);
|
|
int i = 0;
|
|
for (int a = 0; a <= byte.MaxValue; a++)
|
|
{
|
|
for (int b = 0; b <= byte.MaxValue; b++)
|
|
{
|
|
hashPrefixes.Add(StringUtils.FormatAsHexString(new byte[] { (byte)a, (byte)b }));
|
|
i++;
|
|
}
|
|
}
|
|
|
|
hashPrefixes.Shuffle();
|
|
|
|
if (!await AmazonS3Util.DoesS3BucketExistV2Async(_amazonS3, _bucketName))
|
|
{
|
|
yield break;
|
|
}
|
|
|
|
foreach (string hashPrefix in hashPrefixes)
|
|
{
|
|
ListObjectsV2Request request = new ListObjectsV2Request
|
|
{
|
|
BucketName = _bucketName,
|
|
Prefix = hashPrefix,
|
|
MaxKeys = _settings.CurrentValue.PerPrefixMaxKeys
|
|
};
|
|
|
|
ListObjectsV2Response response;
|
|
do
|
|
{
|
|
response = await _amazonS3.ListObjectsV2Async(request, cancellationToken);
|
|
foreach (S3Object obj in response.S3Objects)
|
|
{
|
|
yield return (obj.Key, obj.LastModified);
|
|
}
|
|
|
|
request.ContinuationToken = response.NextContinuationToken;
|
|
} while (response.IsTruncated);
|
|
}
|
|
}
|
|
else
|
|
{
|
|
if (!await AmazonS3Util.DoesS3BucketExistV2Async(_amazonS3, _bucketName))
|
|
{
|
|
yield break;
|
|
}
|
|
|
|
ListObjectsV2Request request = new ListObjectsV2Request
|
|
{
|
|
BucketName = _bucketName
|
|
};
|
|
|
|
ListObjectsV2Response response;
|
|
do
|
|
{
|
|
response = await _amazonS3.ListObjectsV2Async(request, cancellationToken);
|
|
foreach (S3Object obj in response.S3Objects)
|
|
{
|
|
yield return (obj.Key, obj.LastModified);
|
|
}
|
|
|
|
request.ContinuationToken = response.NextContinuationToken;
|
|
} while (response.IsTruncated);
|
|
}
|
|
}
|
|
|
|
public async Task DeleteAsync(string path, CancellationToken cancellationToken)
|
|
{
|
|
await _amazonS3.DeleteObjectAsync(_bucketName, path, cancellationToken);
|
|
}
|
|
|
|
public ValueTask<Uri?> GetReadRedirectAsync(string path)
|
|
{
|
|
return new ValueTask<Uri?>(GetPresignedUrl(path, HttpVerb.GET));
|
|
}
|
|
|
|
public ValueTask<Uri?> GetWriteRedirectAsync(string path)
|
|
{
|
|
return new ValueTask<Uri?>(GetPresignedUrl(path, HttpVerb.PUT));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Helper method to generate a presigned URL for a request
|
|
/// </summary>
|
|
Uri? GetPresignedUrl(string path, HttpVerb verb, int? partNumber = null, string? uploadId = null)
|
|
{
|
|
using TelemetrySpan span = _tracer.StartActiveSpan("s3.BuildPresignedUrl")
|
|
.SetAttribute("Path", path)
|
|
;
|
|
|
|
try
|
|
{
|
|
GetPreSignedUrlRequest signedUrlRequest = new GetPreSignedUrlRequest();
|
|
signedUrlRequest.BucketName = _bucketName;
|
|
signedUrlRequest.Key = path;
|
|
signedUrlRequest.Verb = verb;
|
|
signedUrlRequest.Protocol = _settings.CurrentValue.AssumeHttpForRedirectUri ? Protocol.HTTP : Protocol.HTTPS;
|
|
signedUrlRequest.Expires = DateTime.UtcNow.AddHours(3.0);
|
|
if (partNumber.HasValue)
|
|
{
|
|
signedUrlRequest.PartNumber = partNumber.Value;
|
|
}
|
|
if (uploadId != null)
|
|
{
|
|
signedUrlRequest.UploadId = uploadId;
|
|
}
|
|
|
|
string url = _amazonS3.GetPreSignedURL(signedUrlRequest);
|
|
|
|
return new Uri(url);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogWarning(ex, "Unable to get presigned url for {Path} from S3", path);
|
|
return null;
|
|
}
|
|
}
|
|
|
|
public async Task<BlobMetadata?> GetMetadataAsync(string path)
|
|
{
|
|
try
|
|
{
|
|
GetObjectAttributesResponse? metadata = await _amazonS3.GetObjectAttributesAsync(new GetObjectAttributesRequest
|
|
{
|
|
BucketName = _bucketName, Key = path, ObjectAttributes = new List<ObjectAttributes>() { ObjectAttributes.ObjectSize }
|
|
});
|
|
if (metadata == null)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
return new BlobMetadata(metadata.ObjectSize, metadata.LastModified);
|
|
}
|
|
catch (AmazonS3Exception e)
|
|
{
|
|
// if the object does not exist we get a not found status code
|
|
if (e.StatusCode == HttpStatusCode.NotFound)
|
|
{
|
|
return null;
|
|
}
|
|
|
|
throw;
|
|
}
|
|
catch (XmlException)
|
|
{
|
|
// Multipart uploaded objects can cause invalid xml objects generated in S3
|
|
return null;
|
|
}
|
|
}
|
|
|
|
public async Task<string> StartMultipartUploadAsync(string path)
|
|
{
|
|
InitiateMultipartUploadResponse response = await _amazonS3.InitiateMultipartUploadAsync(new InitiateMultipartUploadRequest {BucketName = _bucketName, Key = path});
|
|
|
|
return response.UploadId;
|
|
}
|
|
|
|
public async Task CompleteMultipartUploadAsync(string path, string uploadId, List<string> partIds)
|
|
{
|
|
Dictionary<int, PartETag> etags = new ();
|
|
|
|
int? partNumberMarker = 0;
|
|
for (int i = 0; i < 10; i++)
|
|
{
|
|
// s3 only allows you to list 1000 parts per request, but supports up to 10_000 parts so we need to potentially list this 10 times to get all parts
|
|
ListPartsRequest listPartsRequest = new() {BucketName = _bucketName, Key = path, UploadId = uploadId, PartNumberMarker = partNumberMarker.ToString()};
|
|
ListPartsResponse partList = await _amazonS3.ListPartsAsync(listPartsRequest);
|
|
foreach (PartDetail? part in partList.Parts)
|
|
{
|
|
etags.Add(part.PartNumber, new PartETag(part.PartNumber, part.ETag));
|
|
}
|
|
if (!partList.IsTruncated)
|
|
{
|
|
break;
|
|
}
|
|
partNumberMarker = partList.NextPartNumberMarker;
|
|
}
|
|
|
|
// check if any part is missing
|
|
List<string> missingPartIds = new();
|
|
foreach (string partId in partIds)
|
|
{
|
|
if (!etags.ContainsKey(int.Parse(partId)))
|
|
{
|
|
missingPartIds.Add(partId);
|
|
}
|
|
}
|
|
|
|
if (missingPartIds.Count > 0)
|
|
{
|
|
throw new MissingMultipartPartsException(missingPartIds!);
|
|
}
|
|
|
|
await _amazonS3.CompleteMultipartUploadAsync(new CompleteMultipartUploadRequest {BucketName = _bucketName, Key = path, UploadId = uploadId, PartETags = etags.Values.ToList()});
|
|
}
|
|
|
|
public async Task PutMultipartPartAsync(string path, string uploadId, string partIdentifier, byte[] data)
|
|
{
|
|
await _amazonS3.UploadPartAsync(new UploadPartRequest { BucketName = _bucketName, Key = path, UploadId = uploadId, PartNumber = int.Parse(partIdentifier), InputStream = new MemoryStream(data)});
|
|
}
|
|
|
|
public Uri? GetWriteRedirectForPart(string path, string uploadId, string partIdentifier)
|
|
{
|
|
Uri? uri = GetPresignedUrl(path, HttpVerb.PUT, partNumber: int.Parse(partIdentifier), uploadId: uploadId);
|
|
|
|
if (uri != null)
|
|
{
|
|
return uri;
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
public async Task RenameMultipartBlobAsync(string blobName, string targetBlobName)
|
|
{
|
|
await CopyBlobAsync(this, blobName, targetBlobName);
|
|
// delete the old multipart object
|
|
await _amazonS3.DeleteObjectAsync(_bucketName, blobName);
|
|
}
|
|
|
|
internal string GetBucketName()
|
|
{
|
|
return _bucketName;
|
|
}
|
|
|
|
public async Task CopyBlobAsync(AmazonStorageBackend source, string sourcePath, string destinationPath)
|
|
{
|
|
if (string.Equals(source._bucketName, _bucketName, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
// copy within the bucket
|
|
if (string.Equals(sourcePath, destinationPath, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
// copy to the same object, this already exists so no need to do any work
|
|
return;
|
|
}
|
|
}
|
|
|
|
GetObjectMetadataResponse metadata = await source._amazonS3.GetObjectMetadataAsync(new GetObjectMetadataRequest {BucketName = source._bucketName, Key = sourcePath, PartNumber = 1} );
|
|
|
|
if (!metadata.PartsCount.HasValue)
|
|
{
|
|
// not a multipart object
|
|
await _amazonS3.CopyObjectAsync(new CopyObjectRequest()
|
|
{
|
|
SourceBucket = source._bucketName, DestinationBucket = _bucketName, SourceKey = sourcePath, DestinationKey = destinationPath,
|
|
});
|
|
|
|
return;
|
|
}
|
|
|
|
InitiateMultipartUploadResponse startMultipartUpload = await _amazonS3.InitiateMultipartUploadAsync(_bucketName, destinationPath);
|
|
string uploadId = startMultipartUpload.UploadId;
|
|
ConcurrentBag<PartETag> parts = new();
|
|
long chunkSize = metadata.ContentLength;
|
|
long objectLength = long.Parse(metadata.ContentRange.Split('/')[1]);
|
|
|
|
await Parallel.ForAsync(1, metadata.PartsCount.Value + 1, async(int i, CancellationToken token) =>
|
|
{
|
|
long startOffset = (i - 1) * chunkSize;
|
|
long endOffset = (i * chunkSize)-1; // last byte is exclusive
|
|
|
|
endOffset = Math.Min(endOffset, objectLength-1);
|
|
CopyPartResponse response = await _amazonS3.CopyPartAsync(new CopyPartRequest()
|
|
{
|
|
SourceBucket = source._bucketName,
|
|
DestinationBucket = _bucketName,
|
|
PartNumber = i,
|
|
SourceKey = sourcePath,
|
|
DestinationKey = destinationPath,
|
|
UploadId = uploadId,
|
|
FirstByte = startOffset,
|
|
LastByte = endOffset,
|
|
}, token);
|
|
parts.Add(new PartETag(i, response.ETag));
|
|
});
|
|
|
|
await _amazonS3.CompleteMultipartUploadAsync(new CompleteMultipartUploadRequest {BucketName = _bucketName, UploadId = uploadId, Key = destinationPath, PartETags = parts.ToList()});
|
|
}
|
|
}
|
|
|
|
public class AmazonS3Factory
|
|
{
|
|
private readonly IServiceProvider _provider;
|
|
|
|
public AmazonS3Factory(IServiceProvider provider)
|
|
{
|
|
_provider = provider;
|
|
}
|
|
|
|
public IAmazonS3 Create(string? region = null)
|
|
{
|
|
S3Settings s3Settings = _provider.GetService<IOptionsMonitor<S3Settings>>()!.CurrentValue;
|
|
AWSCredentials awsCredentials = _provider.GetService<AWSCredentials>()!;
|
|
AWSOptions awsOptions = _provider.GetService<AWSOptions>()!;
|
|
|
|
if (s3Settings.ConnectionString.ToUpper() != "AWS")
|
|
{
|
|
awsOptions.DefaultClientConfig.ServiceURL = s3Settings.ConnectionString;
|
|
}
|
|
|
|
if (region != null)
|
|
{
|
|
awsOptions.Region = RegionEndpoint.GetBySystemName(region);
|
|
}
|
|
|
|
awsOptions.Credentials = awsCredentials;
|
|
|
|
IAmazonS3 serviceClient = awsOptions.CreateServiceClient<IAmazonS3>();
|
|
|
|
if (serviceClient.Config is AmazonS3Config c)
|
|
{
|
|
if (s3Settings.ForceAWSPathStyle)
|
|
{
|
|
c.ForcePathStyle = true;
|
|
// this works around a issue with presigned urls when using minio to emulate S3 due to AWS SDK not defaulting to signature v4 for presigned urls outside of us-east-1
|
|
// see https://github.com/minio/minio/issues/19887
|
|
AWSConfigsS3.UseSignatureVersion4 = true;
|
|
}
|
|
|
|
if (s3Settings.UseArnRegion.HasValue)
|
|
{
|
|
c.UseArnRegion = s3Settings.UseArnRegion.Value;
|
|
}
|
|
}
|
|
else
|
|
{
|
|
throw new NotImplementedException();
|
|
}
|
|
|
|
return serviceClient;
|
|
}
|
|
}
|
|
|
|
public static class BlobIdentifierExtensions
|
|
{
|
|
public static string AsS3Key(this BlobId blobIdentifier)
|
|
{
|
|
string s = blobIdentifier.ToString();
|
|
string prefix = s.Substring(0, 4);
|
|
return $"{prefix}/{s}";
|
|
}
|
|
}
|
|
}
|