Skip to content

Implement configurable message body storage #5646

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
johnsimons wants to merge 4 commits into
base: master
Choose a base branch
from
Open

Implement configurable message body storage #5646

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
a526858
Implement configurable message body storage
johnsimons Jul 23, 2026
b593eeb
Implement Azure Blob storage for message bodies
johnsimons Jul 24, 2026
20611a9
Implement S3 storage for message bodies
johnsimons Jul 24, 2026
f75acd8
Improve message body streaming from cloud storage
johnsimons Jul 24, 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
4 changes: 4 additions & 0 deletions src/Directory.Packages.props
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,11 @@
<ItemGroup Label="Versions for direct package references">
<PackageVersion Include="Autofac" Version="9.3.0" />
<PackageVersion Include="AWSSDK.CloudWatch" Version="4.0.10.5" />
<PackageVersion Include="AWSSDK.S3" Version="4.0.101.4" />
<PackageVersion Include="Azure.Identity" Version="1.21.0" />
<PackageVersion Include="Azure.ResourceManager.Monitor" Version="1.3.1" />
<PackageVersion Include="Azure.ResourceManager.ServiceBus" Version="1.1.0" />
<PackageVersion Include="Azure.Storage.Blobs" Version="12.27.0" />
<PackageVersion Include="ByteSize" Version="2.1.2" />
<PackageVersion Include="Caliburn.Micro" Version="5.0.258" />
<PackageVersion Include="DnsClient" Version="1.8.0" />
Expand Down Expand Up @@ -76,6 +78,8 @@
<PackageVersion Include="PropertyChanging.Fody" Version="1.31.0" />
<PackageVersion Include="PublicApiGenerator" Version="11.5.4" />
<PackageVersion Include="RavenDB.Embedded" Version="6.2.17" />
<PackageVersion Include="Testcontainers.Azurite" Version="4.13.0" />
<PackageVersion Include="Testcontainers.LocalStack" Version="4.13.0" />
<PackageVersion Include="Testcontainers.MsSql" Version="4.13.0" />
<PackageVersion Include="Testcontainers.PostgreSql" Version="4.13.0" />
<PackageVersion Include="ReactiveUI.WPF" Version="22.3.1" />
Expand Down
3 changes: 2 additions & 1 deletion src/ServiceControl.Persistence.EFCore.PostgreSql/PostgreSqlPersistence.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ public void AddPersistence(IServiceCollection services)
{
RegisterSettings(services);
ConfigureDbContext(services);
RegisterDataStores(services);
RegisterDataStores(services, settings);

services.AddSingleton<IIngestionSqlDialect, PostgreSqlIngestionSqlDialect>();
}
Expand All @@ -23,6 +23,7 @@ public void AddInstaller(IServiceCollection services)
ConfigureDbContext(services);

services.AddScoped<IDatabaseMigrator, PostgreSqlDatabaseMigrator>();
RegisterBodyStorageInstaller(services, settings);
}

void RegisterSettings(IServiceCollection services)
Expand Down
48 changes: 48 additions & 0 deletions src/ServiceControl.Persistence.EFCore.PostgreSql/persistence.manifest
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,58 @@
"Name": "ServiceControl/Database/CommandTimeout",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/StorageType",
"Mandatory": true
},
{
"Name": "ServiceControl/MessageBody/StoragePath",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ConnectionString",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ServiceUri",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ManagedIdentityClientId",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/AuthorityHost",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ContainerName",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/BucketName",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/KeyPrefix",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/Region",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/ServiceUrl",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/AccessKeyId",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/SecretAccessKey",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/MinCompressionSize",
"Mandatory": false
Expand Down
3 changes: 2 additions & 1 deletion src/ServiceControl.Persistence.EFCore.SqlServer/SqlServerPersistence.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ public void AddPersistence(IServiceCollection services)
{
RegisterSettings(services);
ConfigureDbContext(services);
RegisterDataStores(services);
RegisterDataStores(services, settings);

services.AddSingleton<IIngestionSqlDialect, SqlServerIngestionSqlDialect>();
}
Expand All @@ -23,6 +23,7 @@ public void AddInstaller(IServiceCollection services)
ConfigureDbContext(services);

services.AddScoped<IDatabaseMigrator, SqlServerDatabaseMigrator>();
RegisterBodyStorageInstaller(services, settings);
}

void RegisterSettings(IServiceCollection services)
Expand Down
48 changes: 48 additions & 0 deletions src/ServiceControl.Persistence.EFCore.SqlServer/persistence.manifest
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,10 +13,58 @@
"Name": "ServiceControl/Database/CommandTimeout",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/StorageType",
"Mandatory": true
},
{
"Name": "ServiceControl/MessageBody/StoragePath",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ConnectionString",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ServiceUri",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ManagedIdentityClientId",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/AuthorityHost",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/Azure/ContainerName",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/BucketName",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/KeyPrefix",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/Region",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/ServiceUrl",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/AccessKeyId",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/S3/SecretAccessKey",
"Mandatory": false
},
{
"Name": "ServiceControl/MessageBody/MinCompressionSize",
"Mandatory": false
Expand Down
41 changes: 39 additions & 2 deletions src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ namespace ServiceControl.Persistence.EFCore.Abstractions;

public abstract class BasePersistence
{
protected static void RegisterDataStores(IServiceCollection services)
protected static void RegisterDataStores(IServiceCollection services, EFPersisterSettings settings)
{
services.AddSingleton(TimeProvider.System);
services.AddSingleton<MinimumRequiredStorageState>();
Expand Down Expand Up @@ -51,6 +51,43 @@ protected static void RegisterDataStores(IServiceCollection services)

services.AddSingleton<ILicensingDataStore, LicensingDataStore>();

services.AddSingleton<IBodyStoragePersistence, FakeBodyStoragePersistence>();
RegisterBodyStorage(services, settings);
}

static void RegisterBodyStorage(IServiceCollection services, EFPersisterSettings settings)
{
switch (settings.BodyStorageType)
{
case BodyStorageType.FileSystem:
services.AddSingleton<IBodyStoragePersistence, FileSystemBodyStoragePersistence>();
break;
case BodyStorageType.AzureBlob:
services.AddSingleton<IBodyStoragePersistence, AzureBlobBodyStoragePersistence>();
break;
case BodyStorageType.S3:
services.AddSingleton<IBodyStoragePersistence, S3BodyStoragePersistence>();
break;
default:
throw new ArgumentOutOfRangeException(nameof(settings), settings.BodyStorageType, "Unknown body storage type.");
}
}

// Only stores needing setup-time provisioning register an installer; SetupCommand skips when none is.
protected static void RegisterBodyStorageInstaller(IServiceCollection services, EFPersisterSettings settings)
{
switch (settings.BodyStorageType)
{
case BodyStorageType.FileSystem:
services.AddScoped<IBodyStorageInstaller, FileSystemBodyStorageInstaller>();
break;
case BodyStorageType.AzureBlob:
services.AddScoped<IBodyStorageInstaller, AzureBlobBodyStorageInstaller>();
break;
case BodyStorageType.S3:
services.AddScoped<IBodyStorageInstaller, S3BodyStorageInstaller>();
break;
default:
throw new ArgumentOutOfRangeException(nameof(settings), settings.BodyStorageType, "Unknown body storage type.");
}
}
}
8 changes: 8 additions & 0 deletions src/ServiceControl.Persistence.EFCore/Abstractions/BodyStorageType.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
namespace ServiceControl.Persistence.EFCore.Abstractions;

public enum BodyStorageType
{
FileSystem,
AzureBlob,
S3
}
110 changes: 109 additions & 1 deletion src/ServiceControl.Persistence.EFCore/Abstractions/EFPersistenceConfigurationBase.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,19 @@ public abstract class EFPersistenceConfigurationBase : IPersistenceConfiguration
{
const string ConnectionStringKey = "Database/ConnectionString";
const string CommandTimeoutKey = "Database/CommandTimeout";
const string BodyStorageTypeKey = "MessageBody/StorageType";
const string MessageBodyStoragePathKey = "MessageBody/StoragePath";
const string AzureConnectionStringKey = "MessageBody/Azure/ConnectionString";
const string AzureServiceUriKey = "MessageBody/Azure/ServiceUri";
const string AzureManagedIdentityClientIdKey = "MessageBody/Azure/ManagedIdentityClientId";
const string AzureAuthorityHostKey = "MessageBody/Azure/AuthorityHost";
const string AzureContainerNameKey = "MessageBody/Azure/ContainerName";
const string S3BucketNameKey = "MessageBody/S3/BucketName";
const string S3KeyPrefixKey = "MessageBody/S3/KeyPrefix";
const string S3RegionKey = "MessageBody/S3/Region";
const string S3ServiceUrlKey = "MessageBody/S3/ServiceUrl";
const string S3AccessKeyIdKey = "MessageBody/S3/AccessKeyId";
const string S3SecretAccessKeyKey = "MessageBody/S3/SecretAccessKey";
const string MinBodySizeForCompressionKey = "MessageBody/MinCompressionSize";
const string MaxBodySizeToStoreKey = "MaxBodySizeToStore";
const string ErrorRetentionPeriodKey = "ErrorRetentionPeriod";
Expand All @@ -19,19 +31,115 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName
var settings = CreateSettings(GetRequiredSetting<string>(settingsRootNamespace, ConnectionStringKey));

settings.CommandTimeout = SettingsReader.Read(settingsRootNamespace, CommandTimeoutKey, EFPersisterSettings.DefaultCommandTimeout);
settings.MessageBodyStoragePath = SettingsReader.Read<string>(settingsRootNamespace, MessageBodyStoragePathKey);
settings.MinBodySizeForCompression = SettingsReader.Read(settingsRootNamespace, MinBodySizeForCompressionKey, EFPersisterSettings.DefaultMinBodySizeForCompression);
settings.MaxBodySizeToStore = ReadMaxBodySizeToStore(settingsRootNamespace);
settings.ErrorRetentionPeriod = GetRequiredSetting<TimeSpan>(settingsRootNamespace, ErrorRetentionPeriodKey);
settings.EnableFullTextSearchOnBodies = SettingsReader.Read(settingsRootNamespace, EnableFullTextSearchOnBodiesKey, true);

ConfigureBodyStorage(settings, settingsRootNamespace);

return settings;
}

public abstract IPersistence Create(PersistenceSettings settings);

protected abstract EFPersisterSettings CreateSettings(string connectionString);

static void ConfigureBodyStorage(EFPersisterSettings settings, SettingsRootNamespace settingsRootNamespace)
{
settings.BodyStorageType = ReadBodyStorageType(settingsRootNamespace);

if (settings.BodyStorageType == BodyStorageType.FileSystem)
{
settings.MessageBodyStoragePath = GetRequiredSetting<string>(settingsRootNamespace, MessageBodyStoragePathKey);
}
else if (settings.BodyStorageType == BodyStorageType.AzureBlob)
{
ConfigureAzureBlob(settings, settingsRootNamespace);
}
else if (settings.BodyStorageType == BodyStorageType.S3)
{
ConfigureS3(settings, settingsRootNamespace);
}
}

static void ConfigureS3(EFPersisterSettings settings, SettingsRootNamespace settingsRootNamespace)
{
settings.S3BucketName = GetRequiredSetting<string>(settingsRootNamespace, S3BucketNameKey);
settings.S3KeyPrefix = SettingsReader.Read(settingsRootNamespace, S3KeyPrefixKey, settings.S3KeyPrefix);
settings.S3Region = SettingsReader.Read<string>(settingsRootNamespace, S3RegionKey);
settings.S3ServiceUrl = SettingsReader.Read<string>(settingsRootNamespace, S3ServiceUrlKey);

var accessKeyId = SettingsReader.Read<string>(settingsRootNamespace, S3AccessKeyIdKey);
var secretAccessKey = SettingsReader.Read<string>(settingsRootNamespace, S3SecretAccessKeyKey);
var hasAccessKeyId = !string.IsNullOrWhiteSpace(accessKeyId);
var hasSecretAccessKey = !string.IsNullOrWhiteSpace(secretAccessKey);

if (hasAccessKeyId != hasSecretAccessKey)
{
throw new Exception($"S3 body storage requires both {S3AccessKeyIdKey} and {S3SecretAccessKeyKey} together, or neither to use the default AWS credential chain (IAM role).");
}

settings.S3AccessKeyId = hasAccessKeyId ? accessKeyId : null;
settings.S3SecretAccessKey = hasSecretAccessKey ? secretAccessKey : null;
}

static void ConfigureAzureBlob(EFPersisterSettings settings, SettingsRootNamespace settingsRootNamespace)
{
var connectionString = SettingsReader.Read<string>(settingsRootNamespace, AzureConnectionStringKey);
var serviceUri = SettingsReader.Read<string>(settingsRootNamespace, AzureServiceUriKey);

var hasConnectionString = !string.IsNullOrWhiteSpace(connectionString);
var hasServiceUri = !string.IsNullOrWhiteSpace(serviceUri);

// A connection string carries shared-key/SAS auth; a service URI uses managed identity. They
// are mutually exclusive, and exactly one must be provided.
if (hasConnectionString == hasServiceUri)
{
throw new Exception($"Azure Blob body storage requires exactly one of {AzureConnectionStringKey} (shared key / SAS) or {AzureServiceUriKey} (managed identity). {(hasConnectionString ? "Both were set." : "Neither was set.")}");
}

settings.AzureBlobConnectionString = hasConnectionString ? connectionString : null;
settings.AzureBlobServiceUri = hasServiceUri ? serviceUri : null;
settings.AzureBlobManagedIdentityClientId = SettingsReader.Read<string>(settingsRootNamespace, AzureManagedIdentityClientIdKey);
settings.AzureBlobAuthorityHost = ReadAzureAuthorityHost(settingsRootNamespace);
settings.AzureBlobContainerName = SettingsReader.Read(settingsRootNamespace, AzureContainerNameKey, settings.AzureBlobContainerName);
}

static string? ReadAzureAuthorityHost(SettingsRootNamespace settingsRootNamespace)
{
var raw = SettingsReader.Read<string>(settingsRootNamespace, AzureAuthorityHostKey);

if (string.IsNullOrWhiteSpace(raw))
{
return null;
}

if (!Uri.TryCreate(raw, UriKind.Absolute, out _))
{
throw new Exception($"Setting {AzureAuthorityHostKey} value '{raw}' is not a valid absolute URI.");
}

return raw;
}

static BodyStorageType ReadBodyStorageType(SettingsRootNamespace settingsRootNamespace)
{
var raw = SettingsReader.Read<string>(settingsRootNamespace, BodyStorageTypeKey);

if (string.IsNullOrWhiteSpace(raw))
{
throw new Exception($"Setting {BodyStorageTypeKey} is required. Valid values: {string.Join(", ", Enum.GetNames<BodyStorageType>())}.");
}

if (!Enum.TryParse<BodyStorageType>(raw, ignoreCase: true, out var value) || !Enum.IsDefined(value))
{
throw new Exception($"Setting {BodyStorageTypeKey} value '{raw}' is not valid. Valid values: {string.Join(", ", Enum.GetNames<BodyStorageType>())}.");
}

return value;
}

static int ReadMaxBodySizeToStore(SettingsRootNamespace settingsRootNamespace)
{
var maxBodySizeToStore = SettingsReader.Read(settingsRootNamespace, MaxBodySizeToStoreKey, EFPersisterSettings.DefaultMaxBodySizeToStore);
Expand Down
12 changes: 12 additions & 0 deletions src/ServiceControl.Persistence.EFCore/Abstractions/EFPersisterSettings.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,19 @@ public abstract class EFPersisterSettings : PersistenceSettings
public required string ConnectionString { get; set; }
public int CommandTimeout { get; set; } = DefaultCommandTimeout;
public TimeSpan ErrorRetentionPeriod { get; set; }
public BodyStorageType BodyStorageType { get; set; }
public string? MessageBodyStoragePath { get; set; }
public string? AzureBlobConnectionString { get; set; }
public string? AzureBlobServiceUri { get; set; }
public string? AzureBlobManagedIdentityClientId { get; set; }
public string? AzureBlobAuthorityHost { get; set; }
public string AzureBlobContainerName { get; set; } = "error-bodies";
public string? S3BucketName { get; set; }
public string S3KeyPrefix { get; set; } = "error-bodies/";
public string? S3Region { get; set; }
public string? S3ServiceUrl { get; set; }
public string? S3AccessKeyId { get; set; }
public string? S3SecretAccessKey { get; set; }
public int MinBodySizeForCompression { get; set; } = DefaultMinBodySizeForCompression;
public int MaxBodySizeToStore { get; set; } = DefaultMaxBodySizeToStore;
public int MaxRetryCount { get; set; } = 5;
Expand Down
Loading
Loading