From 49ff7b2f71fce36d0e6db0aa838031c95fc17e79 Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 21 Aug 2026 12:00:12 +1000 Subject: [PATCH] Let the storage decide whether ingestion batches run in parallel The pipeline was built for several writers and given one. This wires up the settings that reach the rest of it, and puts the decision about whether more than one writer is even allowed where the answer is actually known. Concurrent writers are not a property of the instance, they are a property of the storage, and the two error persisters differ: - EF Core says yes. Its batch writer was built for it, and running several --error-ingestion-only hosts against one database already depends on it. - RavenDB says no. Failed messages are merged by patch scripts that read and rewrite one document with nothing ordering two patches of the same document, and --error-ingestion-only refuses to start on it for the same reason. RavenDB audit says yes, because audit documents are independent and every batch gets its own bulk insert operation. So the audit instance gets parallel writers on the persister it actually ships with, and the error instance gets them on the persisters that were designed for them. SupportsConcurrentBatches on the two unit of work factories carries that, which needs no new registration because both ingestions already take their factory. A storage that says no is held at one writer whatever is configured, and says so in the log only when that overrules a deliberate setting rather than a default. Three settings per instance, ErrorIngestion* and AuditIngestion*, sharing their bounds and their messages through IngestionSettingsReader: - BatchSize, unset meaning the transport's concurrency, which is as many messages as can be waiting to be written at once - MaxParallelWriters, unset meaning the storage decides - BatchTimeout, unset meaning a partial batch is written rather than waited on So an upgrade changes one thing: a storage that takes concurrent batches writes four at a time instead of one. Batch size and the absence of a wait stay as they were, which leaves writer count as the only variable that moved. Enrichers were serialised by the single writer: they run in the synchronous part of processing a message, so one batch ran them one after another on the one loop thread. Several writers is the first time two of them overlap, so both enricher interfaces now say an implementation has to be thread safe. Every enricher in the tree already is, taking its state from the context, and the one that touches anything shared adds to a ConcurrentDictionary. IngestionSettingsReaderTests is NonParallelizable. The assembly runs everything in parallel and every test in the fixture drives the same process-wide environment variable, so one test's teardown cleared what another had just set. --- ...InMemoryAuditIngestionUnitOfWorkFactory.cs | 2 + .../UnitOfWork/RavenAuditUnitOfWorkFactory.cs | 4 + .../IAuditIngestionUnitOfWorkFactory.cs | 7 + ...rovals.PlatformSampleSettings.approved.txt | 3 + .../Auditing/AuditIngestion.cs | 9 +- .../Auditing/IEnrichImportedAuditMessages.cs | 5 + .../Infrastructure/Settings/Settings.cs | 20 +++ .../Ingestion/IngestionSettingsReaderTests.cs | 136 ++++++++++++++++++ .../Ingestion/IngestionSettingsReader.cs | 106 ++++++++++++++ .../EFIngestionUnitOfWorkFactory.cs | 6 + .../RavenIngestionUnitOfWorkFactory.cs | 6 + .../FallbackIngestionUnitOfWorkFactory.cs | 2 + .../UnitOfWork/IIngestionUnitOfWorkFactory.cs | 7 + ...rovals.PlatformSampleSettings.approved.txt | 3 + .../Infrastructure/Settings/Settings.cs | 22 ++- .../Operations/ErrorIngestion.cs | 7 +- .../Operations/IEnrichImportedMessages.cs | 5 + 17 files changed, 346 insertions(+), 4 deletions(-) create mode 100644 src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs create mode 100644 src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs 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);