diff --git a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs index a83db9d32e..d0f0ddf664 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWorkFactory.cs @@ -14,5 +14,7 @@ public ValueTask StartNew(int batchSize, Cancellation } public bool CanIngestMore() => true; + + public bool SupportsConcurrentBatches => false; } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs index 0f256ab17b..84b12d4d3f 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditUnitOfWorkFactory.cs @@ -31,5 +31,9 @@ public async ValueTask StartNew(int batchSize, Cancel } public bool CanIngestMore() => customCheckState.CanIngestMore; + + // Audit documents are independent: nothing merges two of them, and every batch gets its own + // bulk insert operation. + public bool SupportsConcurrentBatches => true; } } diff --git a/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs index 1d3470bdea..f3be8fabd5 100644 --- a/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Audit.Persistence/UnitOfWork/IAuditIngestionUnitOfWorkFactory.cs @@ -7,5 +7,12 @@ public interface IAuditIngestionUnitOfWorkFactory { ValueTask StartNew(int batchSize, CancellationToken cancellationToken = default); //Throws if not enough space or some other problem preventing from writing data bool CanIngestMore(); + + /// + /// Whether several ingestion batches may be written at once. Batches commit in whatever + /// order they finish, so this is only true of a storage whose writes settle the same way + /// however they interleave. + /// + bool SupportsConcurrentBatches { get; } } } \ No newline at end of file diff --git a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index cb9fa30f77..a1901c1f2c 100644 --- a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -63,6 +63,9 @@ "MaximumConcurrencyLevel": null, "ServiceControlQueueAddress": "Particular.ServiceControl", "TimeToRestartAuditIngestionAfterFailure": "00:01:00", + "AuditIngestionBatchSize": null, + "AuditIngestionMaxParallelWriters": null, + "AuditIngestionBatchTimeout": "00:00:00", "EnableFullTextSearchOnBodies": true, "ShutdownTimeout": "00:00:05" } \ No newline at end of file diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs index a1026868e1..533e46f2b4 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs @@ -45,10 +45,15 @@ ILogger logger throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = transportSettings.MaxConcurrency.Value; + MaxBatchSize = settings.AuditIngestionBatchSize ?? transportSettings.MaxConcurrency.Value; pipeline = new IngestionPipeline( - new IngestionPipelineSettings { BatchSize = MaxBatchSize }, + new IngestionPipelineSettings + { + BatchSize = MaxBatchSize, + MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.AuditIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.AuditIngestionMaxParallelWriters), logger), + BatchTimeout = settings.AuditIngestionBatchTimeout + }, IngestBatch, logger); diff --git a/src/ServiceControl.Audit/Auditing/IEnrichImportedAuditMessages.cs b/src/ServiceControl.Audit/Auditing/IEnrichImportedAuditMessages.cs index ec114b6e8e..663317098b 100644 --- a/src/ServiceControl.Audit/Auditing/IEnrichImportedAuditMessages.cs +++ b/src/ServiceControl.Audit/Auditing/IEnrichImportedAuditMessages.cs @@ -1,5 +1,10 @@ namespace ServiceControl.Audit.Auditing { + /// + /// Enriches an audit message as it is ingested. Implementations must be thread safe: ingestion + /// runs several batches at once on storage that allows it, so one instance is called + /// concurrently for messages in different batches. + /// public interface IEnrichImportedAuditMessages { void Enrich(AuditEnricherContext context); diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs index 3203bd349e..e60961c560 100644 --- a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs @@ -9,6 +9,7 @@ using NLog.Common; using NServiceBus.Transport; using ServiceControl.Infrastructure; + using ServiceControl.Infrastructure.Ingestion; using Transports; public class Settings @@ -53,6 +54,9 @@ public Settings(string transportType = null, string persisterType = null, Loggin MaximumConcurrencyLevel = SettingsReader.Read(SettingsRootNamespace, "MaximumConcurrencyLevel"); ServiceControlQueueAddress = SettingsReader.Read(SettingsRootNamespace, "ServiceControlQueueAddress"); TimeToRestartAuditIngestionAfterFailure = GetTimeToRestartAuditIngestionAfterFailure(); + AuditIngestionBatchSize = IngestionSettingsReader.ReadBatchSize(SettingsRootNamespace, nameof(AuditIngestionBatchSize), ValidateConfiguration); + AuditIngestionMaxParallelWriters = IngestionSettingsReader.ReadMaxParallelWriters(SettingsRootNamespace, nameof(AuditIngestionMaxParallelWriters), ValidateConfiguration); + AuditIngestionBatchTimeout = IngestionSettingsReader.ReadBatchTimeout(SettingsRootNamespace, nameof(AuditIngestionBatchTimeout), ValidateConfiguration); EnableFullTextSearchOnBodies = SettingsReader.Read(SettingsRootNamespace, "EnableFullTextSearchOnBodies", true); ShutdownTimeout = SettingsReader.Read(SettingsRootNamespace, "ShutdownTimeout", ShutdownTimeout); @@ -185,6 +189,22 @@ public int MaxBodySizeToStore public TimeSpan TimeToRestartAuditIngestionAfterFailure { get; set; } + /// + /// The most messages one write handles. Null leaves it to the transport's concurrency. + /// + public int? AuditIngestionBatchSize { get; set; } + + /// + /// How many batches are written at once. Null leaves it to the storage, and a storage whose + /// batches are not safe to interleave holds it at one whatever is configured. + /// + public int? AuditIngestionMaxParallelWriters { get; set; } + + /// + /// How long a batch that is not yet full waits for more messages. + /// + public TimeSpan AuditIngestionBatchTimeout { get; set; } + public bool EnableFullTextSearchOnBodies { get; set; } // The default value is set to the maximum allowed time by the most diff --git a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs new file mode 100644 index 0000000000..065fb7c240 --- /dev/null +++ b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs @@ -0,0 +1,136 @@ +namespace ServiceControl.Infrastructure.Tests.Ingestion; + +using System; +using System.Linq; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using ServiceControl.Configuration; +using ServiceControl.Infrastructure.Ingestion; +using ServiceControl.Infrastructure.Tests.Auth; + +// The setting under test is one process-wide environment variable, so these cannot run beside +// each other: one test's teardown clears what another just set. +[TestFixture] +[NonParallelizable] +class IngestionSettingsReaderTests +{ + [TearDown] + public void ClearSetting() => Environment.SetEnvironmentVariable(EnvironmentVariable, null); + + [Test] + public void An_unconfigured_setting_leaves_the_choice_to_the_caller() + { + using (Assert.EnterMultipleScope()) + { + Assert.That(IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: true), Is.Null); + Assert.That(IngestionSettingsReader.ReadMaxParallelWriters(Root, SettingName, validateConfiguration: true), Is.Null); + Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.Zero)); + } + } + + [Test] + public void A_configured_setting_is_read() + { + Configure("250"); + + Assert.That(IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: true), Is.EqualTo(250)); + } + + [Test] + public void A_configured_timeout_is_read() + { + Configure("00:00:00.250"); + + Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.FromMilliseconds(250))); + } + + [TestCase("0")] + [TestCase("1001")] + public void A_batch_size_outside_the_range_is_rejected(string value) + { + Configure(value); + + Assert.That(() => IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: true), Throws.Exception.With.Message.Contains(SettingName)); + } + + [TestCase("0")] + [TestCase("17")] + public void A_writer_count_outside_the_range_is_rejected(string value) + { + Configure(value); + + Assert.That(() => IngestionSettingsReader.ReadMaxParallelWriters(Root, SettingName, validateConfiguration: true), Throws.Exception.With.Message.Contains(SettingName)); + } + + [TestCase("-00:00:01")] + [TestCase("00:00:06")] + public void A_timeout_outside_the_range_is_rejected(string value) + { + Configure(value); + + Assert.That(() => IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Throws.Exception.With.Message.Contains(SettingName)); + } + + [Test] + public void A_timeout_that_is_not_a_TimeSpan_is_rejected_even_without_validation() + { + Configure("soon"); + + Assert.That(() => IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: false), Throws.Exception.With.Message.Contains(SettingName)); + } + + [Test] + public void An_out_of_range_value_is_taken_as_it_stands_when_validation_is_off() + { + Configure("5000"); + + Assert.That(IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: false), Is.EqualTo(5000)); + } + + [Test] + public void A_storage_that_takes_concurrent_batches_gets_the_default_when_nothing_is_configured() => + Assert.That( + IngestionSettingsReader.ResolveMaxParallelWriters(null, storageSupportsConcurrentBatches: true, SettingName, NullLogger.Instance), + Is.EqualTo(IngestionSettingsReader.DefaultMaxParallelWriters)); + + [Test] + public void A_storage_that_takes_concurrent_batches_gets_what_is_configured() => + Assert.That( + IngestionSettingsReader.ResolveMaxParallelWriters(7, storageSupportsConcurrentBatches: true, SettingName, NullLogger.Instance), + Is.EqualTo(7)); + + [Test] + public void A_storage_that_does_not_take_concurrent_batches_is_held_at_one_quietly() + { + using var recorder = new RecordingLoggerProvider(); + + var writers = IngestionSettingsReader.ResolveMaxParallelWriters(null, storageSupportsConcurrentBatches: false, SettingName, recorder.CreateLogger(SettingName)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(writers, Is.EqualTo(1)); + Assert.That(recorder.Entries, Is.Empty, "a default that was never asked for is not worth warning about"); + } + } + + [Test] + public void A_storage_that_does_not_take_concurrent_batches_says_so_when_it_overrules_a_setting() + { + using var recorder = new RecordingLoggerProvider(); + + var writers = IngestionSettingsReader.ResolveMaxParallelWriters(4, storageSupportsConcurrentBatches: false, SettingName, recorder.CreateLogger(SettingName)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(writers, Is.EqualTo(1)); + Assert.That(recorder.Entries.Select(entry => entry.Level), Is.EqualTo(new[] { LogLevel.Warning })); + } + } + + static void Configure(string value) => Environment.SetEnvironmentVariable(EnvironmentVariable, value); + + const string SettingName = "IngestionBatchSetting"; + static readonly SettingsRootNamespace Root = new("IngestionSettingsReaderTests"); + static readonly string EnvironmentVariable = $"{Root}_{SettingName}".ToUpperInvariant(); +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs new file mode 100644 index 0000000000..8ffcdd24a3 --- /dev/null +++ b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs @@ -0,0 +1,106 @@ +namespace ServiceControl.Infrastructure.Ingestion; + +using System; +using Microsoft.Extensions.Logging; +using NLog.Common; +using ServiceControl.Configuration; + +/// +/// Reads and validates the settings that tune an . Each instance +/// names its own after what it ingests, so the caller supplies the name and only the bounds and the +/// messages are shared. +/// +public static class IngestionSettingsReader +{ + /// + /// The most messages one write handles. Absent leaves it to the transport's concurrency, which + /// is as many messages as can be waiting to be written at any one time. + /// + public static int? ReadBatchSize(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration) => + ReadInt(settingsNamespace, name, validateConfiguration, minimum: 1, maximum: MaximumBatchSize); + + /// + /// How many batches are written at once. Absent leaves it to the storage, which is + /// where batches are safe to interleave and one where + /// they are not. + /// + public static int? ReadMaxParallelWriters(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration) => + ReadInt(settingsNamespace, name, validateConfiguration, minimum: 1, maximum: MaximumParallelWriters); + + /// + /// How long a batch that is not yet full waits for more messages. Absent does not wait at all. + /// + public static TimeSpan ReadBatchTimeout(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration) + { + if (!SettingsReader.TryRead(settingsNamespace, name, out var value)) + { + return TimeSpan.Zero; + } + + if (!TimeSpan.TryParse(value, out var timeout)) + { + throw Invalid($"{name} setting is invalid, please make sure it is a TimeSpan."); + } + + if (validateConfiguration && (timeout < TimeSpan.Zero || timeout > MaximumBatchTimeout)) + { + throw Invalid($"{name} setting is invalid, value should be between zero and {MaximumBatchTimeout}."); + } + + return timeout; + } + + /// + /// Settles how many writers a pipeline actually gets. A storage whose batches are not safe to + /// interleave holds it at one whatever is configured, and says so when that overrules a + /// deliberate setting rather than a default. + /// + public static int ResolveMaxParallelWriters(int? configured, bool storageSupportsConcurrentBatches, string settingName, ILogger logger) + { + if (storageSupportsConcurrentBatches) + { + return configured ?? DefaultMaxParallelWriters; + } + + if (configured > 1) + { + logger.LogWarning( + "{SettingName} is set to {ConfiguredWriters}, but the configured storage writes ingestion batches one at a time. One writer is used.", + settingName, configured); + } + + return 1; + } + + static int? ReadInt(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration, int minimum, int maximum) + { + if (!SettingsReader.TryRead(settingsNamespace, name, out var value)) + { + return null; + } + + if (validateConfiguration && (value < minimum || value > maximum)) + { + throw Invalid($"{name} setting is invalid, value should be between {minimum} and {maximum}."); + } + + return value; + } + + // Logged as well as thrown because a bad setting stops the instance before logging is configured + static Exception Invalid(string message) + { + InternalLogger.Fatal(message); + + return new Exception(message); + } + + /// + /// What a storage whose batches are safe to interleave gets when nothing is configured. + /// + public const int DefaultMaxParallelWriters = 4; + + const int MaximumBatchSize = 1000; + const int MaximumParallelWriters = 16; + static readonly TimeSpan MaximumBatchTimeout = TimeSpan.FromSeconds(5); +} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs index 8e2e84bfa6..bd4d4a2d9d 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/UnitOfWork/EFIngestionUnitOfWorkFactory.cs @@ -23,4 +23,10 @@ public ValueTask StartNew(CancellationToken cancellationTo } public bool CanIngestMore() => storageState.CanIngestMore; + + // The batch writer is built for it: upserts guarded by the attempt times so the newer attempt + // wins whichever transaction commits last, inserts that tolerate a competing writer's identical + // row, and a consistent lock order. Running several ingestion hosts against one database + // already relies on all of it. + public bool SupportsConcurrentBatches => true; } diff --git a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenIngestionUnitOfWorkFactory.cs index 1ba74d0e11..0216099bff 100644 --- a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenIngestionUnitOfWorkFactory.cs @@ -15,5 +15,11 @@ public ValueTask StartNew(CancellationToken cancellationTo => new(new RavenIngestionUnitOfWork(sessionProvider, expirationManager, settings)); public bool CanIngestMore() => customCheckState.CanIngestMore; + + // Failed messages are merged by patch scripts that read and rewrite one document, and + // nothing orders two patches of the same document against each other. Error ingestion on + // RavenDB has only ever run one batch at a time, and --error-ingestion-only refuses to + // start on it for the same reason. + public bool SupportsConcurrentBatches => false; } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWorkFactory.cs index c525774b52..e1a738dd27 100644 --- a/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Persistence/UnitOfWork/FallbackIngestionUnitOfWorkFactory.cs @@ -29,5 +29,7 @@ public bool CanIngestMore() { return primary.CanIngestMore() && secondary.CanIngestMore(); } + + public bool SupportsConcurrentBatches => primary.SupportsConcurrentBatches && secondary.SupportsConcurrentBatches; } } \ No newline at end of file diff --git a/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWorkFactory.cs b/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWorkFactory.cs index 5680bf430e..e617e5ae0a 100644 --- a/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWorkFactory.cs +++ b/src/ServiceControl.Persistence/UnitOfWork/IIngestionUnitOfWorkFactory.cs @@ -7,5 +7,12 @@ public interface IIngestionUnitOfWorkFactory { ValueTask StartNew(CancellationToken cancellationToken = default); bool CanIngestMore(); + + /// + /// Whether several ingestion batches may be written at once. Batches commit in whatever + /// order they finish, so this is only true of a storage whose writes settle the same way + /// however they interleave. + /// + bool SupportsConcurrentBatches { get; } } } \ No newline at end of file diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index 203d0facb5..2d2f5fad5a 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -77,6 +77,9 @@ "TransportConnectionString": null, "ProcessRetryBatchesFrequency": "00:00:30", "TimeToRestartErrorIngestionAfterFailure": "00:01:00", + "ErrorIngestionBatchSize": null, + "ErrorIngestionMaxParallelWriters": null, + "ErrorIngestionBatchTimeout": "00:00:00", "MaximumConcurrencyLevel": null, "RetryHistoryDepth": 10, "RemoteInstances": [], diff --git a/src/ServiceControl/Infrastructure/Settings/Settings.cs b/src/ServiceControl/Infrastructure/Settings/Settings.cs index 8455370966..2c1232ec64 100644 --- a/src/ServiceControl/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl/Infrastructure/Settings/Settings.cs @@ -1,4 +1,4 @@ -namespace ServiceBus.Management.Infrastructure.Settings +namespace ServiceBus.Management.Infrastructure.Settings { using System; using System.Collections.Generic; @@ -12,6 +12,7 @@ namespace ServiceBus.Management.Infrastructure.Settings using Particular.ServiceControl; using ServiceControl.Configuration; using ServiceControl.Infrastructure; + using ServiceControl.Infrastructure.Ingestion; using ServiceControl.Infrastructure.Settings; using ServiceControl.Infrastructure.WebApi; using ServiceControl.Persistence; @@ -80,6 +81,9 @@ public Settings( RemoteInstances = GetRemoteInstances().ToArray(); TimeToRestartErrorIngestionAfterFailure = GetTimeToRestartErrorIngestionAfterFailure(); HeartbeatGracePeriod = GetHeartbeatGracePeriod(); + ErrorIngestionBatchSize = IngestionSettingsReader.ReadBatchSize(SettingsRootNamespace, nameof(ErrorIngestionBatchSize), ValidateConfiguration); + ErrorIngestionMaxParallelWriters = IngestionSettingsReader.ReadMaxParallelWriters(SettingsRootNamespace, nameof(ErrorIngestionMaxParallelWriters), ValidateConfiguration); + ErrorIngestionBatchTimeout = IngestionSettingsReader.ReadBatchTimeout(SettingsRootNamespace, nameof(ErrorIngestionBatchTimeout), ValidateConfiguration); DisableExternalIntegrationsPublishing = SettingsReader.Read(SettingsRootNamespace, "DisableExternalIntegrationsPublishing", false); TrackInstancesInitialValue = SettingsReader.Read(SettingsRootNamespace, "TrackInstancesInitialValue", true); ShutdownTimeout = SettingsReader.Read(SettingsRootNamespace, "ShutdownTimeout", ShutdownTimeout); @@ -203,6 +207,22 @@ public string InstanceId public string TransportConnectionString { get; set; } public TimeSpan ProcessRetryBatchesFrequency { get; set; } public TimeSpan TimeToRestartErrorIngestionAfterFailure { get; set; } + + /// + /// The most messages one write handles. Null leaves it to the transport's concurrency. + /// + public int? ErrorIngestionBatchSize { get; set; } + + /// + /// How many batches are written at once. Null leaves it to the storage, and a storage whose + /// batches are not safe to interleave holds it at one whatever is configured. + /// + public int? ErrorIngestionMaxParallelWriters { get; set; } + + /// + /// How long a batch that is not yet full waits for more messages. + /// + public TimeSpan ErrorIngestionBatchTimeout { get; set; } public int? MaximumConcurrencyLevel { get; set; } public int RetryHistoryDepth { get; set; } diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index 456a2bad8a..107ca9550c 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -51,7 +51,12 @@ public ErrorIngestion( } pipeline = new IngestionPipeline( - new IngestionPipelineSettings { BatchSize = transportSettings.MaxConcurrency.Value }, + new IngestionPipelineSettings + { + BatchSize = settings.ErrorIngestionBatchSize ?? transportSettings.MaxConcurrency.Value, + MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.ErrorIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.ErrorIngestionMaxParallelWriters), logger), + BatchTimeout = settings.ErrorIngestionBatchTimeout + }, IngestBatch, logger); diff --git a/src/ServiceControl/Operations/IEnrichImportedMessages.cs b/src/ServiceControl/Operations/IEnrichImportedMessages.cs index 4b3efb63bb..6040f0b450 100644 --- a/src/ServiceControl/Operations/IEnrichImportedMessages.cs +++ b/src/ServiceControl/Operations/IEnrichImportedMessages.cs @@ -1,5 +1,10 @@ namespace ServiceControl.Operations { + /// + /// Enriches an error message as it is ingested. Implementations must be thread safe: ingestion + /// runs several batches at once on storage that allows it, so one instance is called + /// concurrently for messages in different batches. + /// public interface IEnrichImportedErrorMessages { void Enrich(ErrorEnricherContext context);