Skip to content

Upload S3 files concurrently instead of one directory call #4238

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Mpdreamz wants to merge 4 commits into
base: main
Choose a base branch
from
Open

Upload S3 files concurrently instead of one directory call #4238

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
a98264e
Upload S3 files concurrently instead of one directory call
Mpdreamz Sep 29, 2026
562d5c7
Tests: prevent OTel activity bleed between concurrent Apply() tests
Mpdreamz Sep 29, 2026
6bf0f6f
Deploying: treat S3 per-key delete failures as errors regardless of H…
Mpdreamz Sep 29, 2026
de5bb04
Merge main into the parallel S3 sync branch
Mpdreamz Nov 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions src/services/Elastic.Documentation.Deploying/IncrementalDeployService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
// See the LICENSE file in the project root for more information

using Actions.Core.Services;
using Amazon.Runtime;
using Amazon.S3;
using Amazon.S3.Transfer;
using Elastic.Documentation.Deploying.Synchronization;
Expand All @@ -22,7 +23,11 @@ public class IncrementalDeployService(
) : IService
{
private readonly ILogger _logger = logFactory.CreateLogger<IncrementalDeployService>();
private readonly IAmazonS3 _s3 = s3Client ?? new AmazonS3Client();

// Standard mode retries throttling/5xx with jittered backoff; higher upload concurrency means more
// concurrent requests hitting S3, so let the SDK absorb transient failures instead of failing files outright.
private readonly IAmazonS3 _s3 = s3Client
?? new AmazonS3Client(new AmazonS3Config { RetryMode = RequestRetryMode.Standard, MaxErrorRetry = 5 });

public async Task<bool> Plan(
IDiagnosticsCollector collector,
Expand Down Expand Up @@ -105,8 +110,7 @@ Cancel ctx
return false;
}
var applier = new AwsS3SyncApplyStrategy(logFactory, _s3, xfer, s3BucketName, context, collector);
await applier.Apply(plan, ctx);
return true;
return await applier.Apply(plan, ctx);
}

private void LogPlanSummary(SyncPlan plan)
Expand Down
193 changes: 113 additions & 80 deletions src/services/Elastic.Documentation.Deploying/Synchronization/AwsS3SyncApplyStrategy.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,12 @@

using System.Diagnostics;
using System.Diagnostics.Metrics;
using System.Threading;
using Amazon.S3;
using Amazon.S3.Model;
using Amazon.S3.Transfer;
using Elastic.Documentation.Diagnostics;
using Elastic.Documentation.Integrations.S3;
using Elastic.Documentation.ServiceDefaults.Telemetry;
using Microsoft.Extensions.Logging;

Expand All @@ -26,6 +28,12 @@ IDiagnosticsCollector collector

// Meter for OpenTelemetry metrics
private static readonly Meter SyncMeter = new(TelemetryConstants.AssemblerSyncInstrumentationName);
private const int DefaultUploadConcurrency = 32;
private const int MaxUploadConcurrency = 128;
private const int DefaultDeleteConcurrency = 4;
private const int MaxDeleteConcurrency = 16;
private const string UploadConcurrencyEnvironmentVariable = "DOCS_SYNC_UPLOAD_CONCURRENCY";
private const string DeleteConcurrencyEnvironmentVariable = "DOCS_SYNC_DELETE_CONCURRENCY";

// Deployment-level metrics (histograms for distribution analysis, counters for totals)
// Note: Histograms require delta temporality to work with Elasticsearch
Expand Down Expand Up @@ -86,17 +94,18 @@ IDiagnosticsCollector collector

private readonly ILogger<AwsS3SyncApplyStrategy> _logger = logFactory.CreateLogger<AwsS3SyncApplyStrategy>();

private void DisplayProgress(object? sender, UploadDirectoryProgressArgs args) => LogProgress(_logger, args);
private void DisplayUploadProgress(object? sender, UploadProgressArgs args) => LogUploadProgress(_logger, args);

[LoggerMessage(EventId = 2, Level = LogLevel.Debug, Message = "{Args}")]
private static partial void LogProgress(ILogger logger, UploadDirectoryProgressArgs args);
[LoggerMessage(EventId = 4, Level = LogLevel.Debug, Message = "{Args}")]
private static partial void LogUploadProgress(ILogger logger, UploadProgressArgs args);

[LoggerMessage(EventId = 3, Level = LogLevel.Information, Message = "File operation: {Operation} | Path: {FilePath} | Size: {FileSize} bytes")]
private static partial void LogFileOperation(ILogger logger, string operation, string filePath, long fileSize);

public async Task Apply(SyncPlan plan, Cancel ctx = default)
public async Task<bool> Apply(SyncPlan plan, Cancel ctx = default)
{
var sw = Stopwatch.StartNew();
var errorsBeforeApply = collector.Errors;

using var applyActivity = ApplyStrategyActivitySource.StartActivity("sync apply", ActivityKind.Client);
if (Environment.GetEnvironmentVariable("GITHUB_ACTIONS") == "true")
Expand Down Expand Up @@ -153,15 +162,21 @@ public async Task Apply(SyncPlan plan, Cancel ctx = default)

// Record sync duration (both histogram for distribution and counter for total)
SyncDurationHistogram.Record(sw.Elapsed.TotalSeconds);

return collector.Errors == errorsBeforeApply;
}

private async Task Upload(SyncPlan plan, Cancel ctx)
{
var sw = Stopwatch.StartNew();
var uploadedCount = 0;
var uploadRequests = plan.AddRequests.Cast<UploadRequest>().Concat(plan.UpdateRequests).ToList();
var uploadConcurrency = GetConcurrency(UploadConcurrencyEnvironmentVariable, DefaultUploadConcurrency, MaxUploadConcurrency);

// Always create activity span (even if 0 files) for consistent tracing
using var uploadActivity = ApplyStrategyActivitySource.StartActivity("upload files", ActivityKind.Client);
_ = uploadActivity?.SetTag("docs.sync.upload.count", uploadRequests.Count);
_ = uploadActivity?.SetTag("docs.sync.upload.concurrency", uploadConcurrency);

if (uploadRequests.Count > 0)
{
Expand All @@ -171,16 +186,19 @@ private async Task Upload(SyncPlan plan, Cancel ctx)
_logger.LogInformation("Starting to process {AddCount} new files and {UpdateCount} updated files", addCount, updateCount);

var addPaths = plan.AddRequests.Select(r => r.LocalPath).ToHashSet();
var tempDir = Path.Join(context.WriteFileSystem.Path.GetTempPath(), context.WriteFileSystem.Path.GetRandomFileName());
_ = context.WriteFileSystem.Directory.CreateDirectory(tempDir);
// Files with an explicit Content-Type are staged in sibling directories, one per type, so each
// directory upload carries its own type. The SDK shares one Headers collection across all per-file
// requests of a directory upload, so setting ContentType per file would leak it onto other files.
var typedDirs = new Dictionary<string, string>();
try
_logger.LogInformation(
"Uploading {Count} files to S3 bucket {BucketName} with concurrency {Concurrency}",
uploadRequests.Count,
bucketName,
uploadConcurrency
);
await Parallel.ForEachAsync(uploadRequests, new ParallelOptions
{
_logger.LogInformation("Copying {Count} files to temp directory", uploadRequests.Count);
foreach (var upload in uploadRequests)
CancellationToken = ctx,
MaxDegreeOfParallelism = uploadConcurrency
}, async (upload, token) =>
{
try
{
var operation = addPaths.Contains(upload.LocalPath) ? "add" : "update";
var fileSize = context.WriteFileSystem.FileInfo.New(upload.LocalPath).Length;
Expand All @@ -191,60 +209,42 @@ private async Task Upload(SyncPlan plan, Cancel ctx)
FilesByExtensionCounter.Add(1, new("operation", operation), new("extension", extension));
LogFileOperation(_logger, operation, upload.DestinationPath, fileSize);

var stagingDir = tempDir;
if (SpecContentType(upload.DestinationPath) is { } specType)
{
if (!typedDirs.TryGetValue(specType, out stagingDir))
typedDirs[specType] = stagingDir = $"{tempDir}-{typedDirs.Count}";
}

var destPath = context.WriteFileSystem.Path.Join(stagingDir, upload.DestinationPath);
var destDirPath = context.WriteFileSystem.Path.GetDirectoryName(destPath)!;
_ = context.WriteFileSystem.Directory.CreateDirectory(destDirPath);
context.WriteFileSystem.File.Copy(upload.LocalPath, destPath);
}
_logger.LogInformation("Uploading {Count} files to S3 bucket {BucketName}", uploadRequests.Count, bucketName);
var uploads = typedDirs.Select(kv => (Directory: kv.Value, ContentType: (string?)kv.Key)).Prepend(
(Directory: tempDir, ContentType: null)
);
foreach (var (directory, contentType) in uploads)
{
if (
!context.WriteFileSystem.Directory.Exists(directory)
|| !context.WriteFileSystem.Directory.EnumerateFiles(directory, "*", SearchOption.AllDirectories).Any()
)
continue;
var directoryRequest = new TransferUtilityUploadDirectoryRequest
var request = new TransferUtilityUploadRequest
{
BucketName = bucketName,
Directory = directory,
SearchPattern = "*",
SearchOption = SearchOption.AllDirectories,
UploadFilesConcurrently = true
FilePath = upload.LocalPath,
Key = upload.DestinationPath,
PartSize = S3EtagCalculator.PartSize
};
if (contentType is not null)
directoryRequest.ContentType = contentType;
directoryRequest.UploadDirectoryProgressEvent += DisplayProgress;
_logger.LogDebug("Starting directory upload from {Directory}", directory);
await transferUtility.UploadDirectoryAsync(directoryRequest, ctx);
// Per-file requests own their headers, so the type cannot leak onto other uploads.
if (SpecContentType(upload.DestinationPath) is { } specType)
request.ContentType = specType;
request.UploadProgressEvent += DisplayUploadProgress;
await transferUtility.UploadAsync(request, token);
_ = Interlocked.Increment(ref uploadedCount);
}
_logger.LogInformation(
"Successfully uploaded {Count} files ({AddCount} added, {UpdateCount} updated)",
uploadRequests.Count,
addCount,
updateCount
);
}
finally
{
// Clean up temp directories
foreach (var directory in typedDirs.Values.Prepend(tempDir))
catch (Exception ex) when (ex is not OperationCanceledException)

@github-actions github-actions Bot Sep 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[HIGH] Upload failures are swallowed, then deletes still execute

This catch turns per-file upload failures into diagnostics and continues, but Apply still proceeds to the delete phase. That can produce destructive partial deploys: if a plan includes add new-path + delete old-path (rename/move case), a failed upload of new-path can be followed by successful deletion of old-path, leaving content missing in S3.

Please gate deletes on upload success (for example, track upload-phase failures and return early before Delete when any upload failed).

{
if (context.WriteFileSystem.Directory.Exists(directory))
context.WriteFileSystem.Directory.Delete(directory, true);
_logger.LogError(
ex,
"Failed to upload {LocalPath} to s3://{BucketName}/{DestinationPath}",
upload.LocalPath,
bucketName,
upload.DestinationPath
);
collector.EmitError(upload.LocalPath, $"Failed to upload to s3://{bucketName}/{upload.DestinationPath}", ex);
}
}
});
_logger.LogInformation(
"Finished uploading {UploadedCount}/{Count} files ({AddCount} added, {UpdateCount} updated) in {ElapsedMs}ms",
uploadedCount,
uploadRequests.Count,
addCount,
updateCount,
sw.ElapsedMilliseconds
);
}
_ = uploadActivity?.SetTag("docs.sync.upload.duration_ms", sw.Elapsed.TotalMilliseconds);
}

/// <summary>
Expand All @@ -261,16 +261,24 @@ private async Task Upload(SyncPlan plan, Cancel ctx)

private async Task Delete(SyncPlan plan, Cancel ctx)
{
var sw = Stopwatch.StartNew();
var deleteCount = 0;
var deleteRequests = plan.DeleteRequests;
var deleteConcurrency = GetConcurrency(DeleteConcurrencyEnvironmentVariable, DefaultDeleteConcurrency, MaxDeleteConcurrency);

// Always create activity span (even if 0 files) for consistent tracing
using var deleteActivity = ApplyStrategyActivitySource.StartActivity("delete files", ActivityKind.Client);
_ = deleteActivity?.SetTag("docs.sync.delete.count", deleteRequests.Count);
_ = deleteActivity?.SetTag("docs.sync.delete.concurrency", deleteConcurrency);

if (deleteRequests.Count > 0)
{
_logger.LogInformation("Starting to delete {Count} files from S3 bucket {BucketName}", deleteRequests.Count, bucketName);
_logger.LogInformation(
"Starting to delete {Count} files from S3 bucket {BucketName} with concurrency {Concurrency}",
deleteRequests.Count,
bucketName,
deleteConcurrency
);

// Emit file-level metrics (low cardinality) and logs for each file
foreach (var delete in deleteRequests)
Expand All @@ -279,45 +287,70 @@ private async Task Delete(SyncPlan plan, Cancel ctx)

// Record by extension (low cardinality)
if (!string.IsNullOrEmpty(extension))
{
FilesByExtensionCounter.Add(1, new("operation", "delete"), new("extension", extension));
}

// Log individual file operations for detailed analysis
LogFileOperation(_logger, "delete", delete.DestinationPath, 0);
}

// Process deletes in batches of 1000 (AWS S3 limit)
foreach (var batch in deleteRequests.Chunk(1000))
// Process deletes in batches of 1000 (AWS S3 limit), parallelised across batches
await Parallel.ForEachAsync(deleteRequests.Chunk(1000), new ParallelOptions
{
CancellationToken = ctx,
MaxDegreeOfParallelism = deleteConcurrency
}, async (batch, token) =>
{
var deleteObjectsRequest = new DeleteObjectsRequest
{
BucketName = bucketName,
Objects = batch.Select(d => new KeyVersion { Key = d.DestinationPath }).ToList()
};
var response = await s3Client.DeleteObjectsAsync(deleteObjectsRequest, ctx);
if (response.HttpStatusCode != System.Net.HttpStatusCode.OK)
try
{
_logger.LogError("Delete batch failed with status code {StatusCode}", response.HttpStatusCode);
foreach (var error in response.DeleteErrors)
var response = await s3Client.DeleteObjectsAsync(deleteObjectsRequest, token);
// S3 returns 200 even for partial failures; per-key errors live in DeleteErrors.
var errors = response.DeleteErrors ?? [];
foreach (var error in errors)
{
_logger.LogError("Failed to delete {Key}: {Message}", error.Key, error.Message);
collector.EmitError(error.Key, $"Failed to delete: {error.Message}");
}
if (response.HttpStatusCode != System.Net.HttpStatusCode.OK && errors.Count == 0)
collector.EmitGlobalError($"Delete batch failed with status code {response.HttpStatusCode}");

var successCount = batch.Length - errors.Count;
if (successCount > 0)
{
var currentCount = Interlocked.Add(ref deleteCount, successCount);
_logger.LogInformation(
"Deleted {BatchCount} files ({CurrentCount}/{TotalCount})",
successCount,
currentCount,
deleteRequests.Count
);
}
}
else
catch (Exception ex) when (ex is not OperationCanceledException)
{
deleteCount += batch.Length;
_logger.LogInformation(
"Deleted {BatchCount} files ({CurrentCount}/{TotalCount})",
batch.Length,
deleteCount,
deleteRequests.Count
);
_logger.LogError(ex, "Failed to delete batch from S3 bucket {BucketName}", bucketName);
foreach (var delete in batch)
collector.EmitError(delete.DestinationPath, "Failed to delete from S3", ex);
}
}

_logger.LogInformation("Successfully deleted {Count} files", deleteCount);
});

_logger.LogInformation(
"Finished deleting {DeletedCount}/{Count} files in {ElapsedMs}ms",
deleteCount,
deleteRequests.Count,
sw.ElapsedMilliseconds
);
}
_ = deleteActivity?.SetTag("docs.sync.delete.duration_ms", sw.Elapsed.TotalMilliseconds);
}

private static int GetConcurrency(string environmentVariable, int defaultValue, int maxValue)
{
var configured = Environment.GetEnvironmentVariable(environmentVariable);
return int.TryParse(configured, out var parsed) && parsed > 0 ? Math.Min(parsed, maxValue) : defaultValue;
}
}
3 changes: 2 additions & 1 deletion src/services/Elastic.Documentation.Deploying/Synchronization/DocsSync.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,8 @@ public record PlanValidationResult(bool Valid, float DeleteRatio, float DeleteTh

public interface IDocsSyncApplyStrategy
{
Task Apply(SyncPlan plan, Cancel ctx = default);
/// <returns>false if any file failed to upload or delete.</returns>
Task<bool> Apply(SyncPlan plan, Cancel ctx = default);
}

public record SyncRequest;
Expand Down
Loading
Loading