diff --git a/docs/ingestion-pipeline.md b/docs/ingestion-pipeline.md index d747db318d..2b3d2e358a 100644 --- a/docs/ingestion-pipeline.md +++ b/docs/ingestion-pipeline.md @@ -46,9 +46,9 @@ Each instance reads its settings from its own settings root | Setting | Default | Range | What it does | | --- | --- | --- | --- | -| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | the transport's `MaximumConcurrencyLevel` | 1 to 1000 | The most messages one write handles | -| `ErrorIngestionMaxParallelWriters` / `AuditIngestionMaxParallelWriters` | the storage decides, see below | 1 to 16 | How many batches are written at once | -| `ErrorIngestionBatchTimeout` / `AuditIngestionBatchTimeout` | `00:00:00` | 0 to 5 seconds | How long a batch that is not yet full waits for more messages | +| `ErrorIngestionBatchSize` / `AuditIngestionBatchSize` | the transport's `MaximumConcurrencyLevel` divided by the writers, at least 1 | 1 to 1000 | The most messages one write handles | +| `ErrorIngestionMaxParallelWriters` / `AuditIngestionMaxParallelWriters` | `2` where the storage takes concurrent batches, otherwise `1`, see below | 1 to 16 | How many batches are written at once | +| `ErrorIngestionBatchTimeout` / `AuditIngestionBatchTimeout` | `00:00:00.100` | 0 to 5 seconds | How long a batch that is not yet full waits for more messages | As environment variables: @@ -64,21 +64,13 @@ SERVICECONTROL_AUDIT_AuditIngestionBatchTimeout=00:00:00.100 ### Batch size -The default, the transport's concurrency, is also the ceiling. A receive does not return until the -batch carrying its message has been written, so the transport holds every one of its concurrency -slots open and no more than that many messages can ever be waiting. Setting a batch size above the -transport's concurrency is therefore unreachable, with or without a batch timeout: the batch never -fills, and a timeout only delays what has already arrived. Larger batches come from raising -`MaximumConcurrencyLevel`, and this setting is what holds writes below it when a storage is happier -with smaller ones. +The default batch size is the transport's concurrency divided by the number of writers, using integer division and never less than 1. The writer count is the one the pipeline actually gets, so a storage that holds ingestion to one writer gets a batch as large as the concurrency. When the concurrency is below the writer count, the batch size is 1. A receive does not return until its batch has been written, so no more than the concurrency can wait. A configured batch size is used as given. One above the transport's concurrency never fills, and a timeout only delays messages already received. + +Without `MaximumConcurrencyLevel`, the error instance receives with a concurrency of 10. The audit instance defaults to 32, except SQL Server and PostgreSQL transports, which default to 10. In the error instance, this shared setting also controls the hosted endpoint, retry forwarding, and throughput collection receivers. To allow larger batches, raise `MaximumConcurrencyLevel`. ### Batch timeout -Zero means a partial batch is written without waiting, which is what the ingestion did before -the setting existed. A non-zero value trades latency for fewer, larger writes: at volume it costs -nothing, because a full batch never waits, and at a trickle it delays each message by up to the -timeout. Start at 100ms if a storage is clearly happier with larger batches. It cannot make a batch -larger than the transport's concurrency, only fuller. +The default is 100 ms. An explicit zero writes partial batches without waiting. A nonzero timeout trades latency for fewer, larger writes: full batches never wait, but partial batches wait up to the timeout. The default can therefore add up to 100 ms of latency at low message volumes. A timeout cannot make a batch larger than the transport's concurrency, only fuller. ### Parallel writers @@ -100,9 +92,10 @@ someone actually asked for rather than a default. | Error, SQL Server and PostgreSQL | yes | The batch writer was 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 `--error-ingestion-only` hosts against one database already depends on all of it. | | Error, RavenDB | no | 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-only` refuses to start on RavenDB for the same reason. | | Audit, RavenDB | yes | Audit documents are independent. Nothing merges two of them, and every batch gets its own bulk insert operation. | +| Audit, SQL Server and PostgreSQL | yes | The audit unit of work factory supports concurrent batches. | | Audit, in memory | no | Test storage only. | -A storage that says yes gets four writers by default. +Error ingestion defaults to two writers on SQL Server and PostgreSQL and one on RavenDB. Audit ingestion defaults to two writers on all three providers. In-memory test storage uses one writer. ### The ordering that concurrency does not excuse diff --git a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index 44ef85404e..f0948c181f 100644 --- a/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.Audit.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -64,7 +64,7 @@ "TimeToRestartAuditIngestionAfterFailure": "00:01:00", "AuditIngestionBatchSize": null, "AuditIngestionMaxParallelWriters": null, - "AuditIngestionBatchTimeout": "00:00:00", + "AuditIngestionBatchTimeout": "00:00:00.1000000", "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 533e46f2b4..fece955a25 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs @@ -45,13 +45,14 @@ ILogger logger throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = settings.AuditIngestionBatchSize ?? transportSettings.MaxConcurrency.Value; + var writers = IngestionSettingsReader.ResolveMaxParallelWriters(settings.AuditIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.AuditIngestionMaxParallelWriters), logger); + MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.AuditIngestionBatchSize, transportSettings.MaxConcurrency.Value, writers); pipeline = new IngestionPipeline( new IngestionPipelineSettings { BatchSize = MaxBatchSize, - MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.AuditIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.AuditIngestionMaxParallelWriters), logger), + MaxWriters = writers, BatchTimeout = settings.AuditIngestionBatchTimeout }, IngestBatch, diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs index 7a6b8d978f..da096a3720 100644 --- a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs @@ -189,7 +189,7 @@ public int MaxBodySizeToStore public TimeSpan TimeToRestartAuditIngestionAfterFailure { get; set; } /// - /// The most messages one write handles. Null leaves it to the transport's concurrency. + /// The most messages one write handles. Null splits the transport's concurrency across the writers. /// public int? AuditIngestionBatchSize { get; set; } @@ -200,7 +200,7 @@ public int MaxBodySizeToStore public int? AuditIngestionMaxParallelWriters { get; set; } /// - /// How long a batch that is not yet full waits for more messages. + /// How long a batch that is not yet full waits for more messages. Unset is 100 ms; zero does not wait. /// public TimeSpan AuditIngestionBatchTimeout { get; set; } diff --git a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs index 065fb7c240..a352cf0c64 100644 --- a/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs +++ b/src/ServiceControl.Infrastructure.Tests/Ingestion/IngestionSettingsReaderTests.cs @@ -19,13 +19,13 @@ class IngestionSettingsReaderTests public void ClearSetting() => Environment.SetEnvironmentVariable(EnvironmentVariable, null); [Test] - public void An_unconfigured_setting_leaves_the_choice_to_the_caller() + public void An_unconfigured_setting_is_left_to_the_caller_or_the_default() { 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)); + Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.FromMilliseconds(100))); } } @@ -37,6 +37,28 @@ public void A_configured_setting_is_read() Assert.That(IngestionSettingsReader.ReadBatchSize(Root, SettingName, validateConfiguration: true), Is.EqualTo(250)); } + [Test] + public void A_configured_zero_timeout_is_kept() + { + Configure("00:00:00"); + + Assert.That(IngestionSettingsReader.ReadBatchTimeout(Root, SettingName, validateConfiguration: true), Is.EqualTo(TimeSpan.Zero)); + } + + [TestCase(10, 1, 10)] + [TestCase(10, 4, 2)] + [TestCase(32, 4, 8)] + [TestCase(10, 3, 3)] + [TestCase(3, 4, 1)] + [TestCase(1, 4, 1)] + public void An_unconfigured_batch_size_divides_the_transport_concurrency_across_the_writers(int concurrency, int writers, int expected) => + Assert.That(IngestionSettingsReader.ResolveBatchSize(null, concurrency, writers), Is.EqualTo(expected)); + + [TestCase(10, 1)] + [TestCase(100, 4)] + public void A_configured_batch_size_is_kept(int concurrency, int writers) => + Assert.That(IngestionSettingsReader.ResolveBatchSize(50, concurrency, writers), Is.EqualTo(50)); + [Test] public void A_configured_timeout_is_read() { @@ -92,7 +114,7 @@ public void An_out_of_range_value_is_taken_as_it_stands_when_validation_is_off() 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)); + Is.EqualTo(2)); [Test] public void A_storage_that_takes_concurrent_batches_gets_what_is_configured() => diff --git a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs index 8ffcdd24a3..1410fdfeae 100644 --- a/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs +++ b/src/ServiceControl.Infrastructure/Ingestion/IngestionSettingsReader.cs @@ -13,8 +13,7 @@ namespace ServiceControl.Infrastructure.Ingestion; 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. + /// The most messages one write handles. Absent leaves it to . /// public static int? ReadBatchSize(SettingsRootNamespace settingsNamespace, string name, bool validateConfiguration) => ReadInt(settingsNamespace, name, validateConfiguration, minimum: 1, maximum: MaximumBatchSize); @@ -28,13 +27,14 @@ public static class IngestionSettingsReader 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. + /// How long a batch that is not yet full waits for more messages. Absent waits + /// ; an explicit zero 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; + return DefaultBatchTimeout; } if (!TimeSpan.TryParse(value, out var timeout)) @@ -50,6 +50,14 @@ public static TimeSpan ReadBatchTimeout(SettingsRootNamespace settingsNamespace, return timeout; } + /// + /// Settles how many messages one write handles. The default splits the transport's concurrency + /// across the writers, so that each can fill a batch while the others are writing. + /// is the resolved writer count, see . + /// + public static int ResolveBatchSize(int? configured, int transportMaxConcurrency, int writers) => + configured ?? Math.Max(1, Math.Min(transportMaxConcurrency, transportMaxConcurrency / writers)); + /// /// 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 @@ -98,7 +106,9 @@ static Exception Invalid(string message) /// /// What a storage whose batches are safe to interleave gets when nothing is configured. /// - public const int DefaultMaxParallelWriters = 4; + public const int DefaultMaxParallelWriters = 2; + + public static readonly TimeSpan DefaultBatchTimeout = TimeSpan.FromMilliseconds(100); const int MaximumBatchSize = 1000; const int MaximumParallelWriters = 16; diff --git a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt index abd4c98313..b7303ff3a8 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -79,7 +79,7 @@ "TimeToRestartErrorIngestionAfterFailure": "00:01:00", "ErrorIngestionBatchSize": null, "ErrorIngestionMaxParallelWriters": null, - "ErrorIngestionBatchTimeout": "00:00:00", + "ErrorIngestionBatchTimeout": "00:00:00.1000000", "MaximumConcurrencyLevel": null, "RetryHistoryDepth": 10, "RemoteInstances": [], diff --git a/src/ServiceControl/Infrastructure/Settings/Settings.cs b/src/ServiceControl/Infrastructure/Settings/Settings.cs index fad7491699..c601c07b57 100644 --- a/src/ServiceControl/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl/Infrastructure/Settings/Settings.cs @@ -210,7 +210,7 @@ public string InstanceId public TimeSpan TimeToRestartErrorIngestionAfterFailure { get; set; } /// - /// The most messages one write handles. Null leaves it to the transport's concurrency. + /// The most messages one write handles. Null splits the transport's concurrency across the writers. /// public int? ErrorIngestionBatchSize { get; set; } @@ -221,7 +221,7 @@ public string InstanceId public int? ErrorIngestionMaxParallelWriters { get; set; } /// - /// How long a batch that is not yet full waits for more messages. + /// How long a batch that is not yet full waits for more messages. Unset is 100 ms; zero does not wait. /// public TimeSpan ErrorIngestionBatchTimeout { get; set; } public int? MaximumConcurrencyLevel { get; set; } diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index d23623c6d4..6bf57e1e38 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -45,13 +45,14 @@ public ErrorIngestion( throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } - MaxBatchSize = settings.ErrorIngestionBatchSize ?? transportSettings.MaxConcurrency.Value; + var writers = IngestionSettingsReader.ResolveMaxParallelWriters(settings.ErrorIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.ErrorIngestionMaxParallelWriters), logger); + MaxBatchSize = IngestionSettingsReader.ResolveBatchSize(settings.ErrorIngestionBatchSize, transportSettings.MaxConcurrency.Value, writers); pipeline = new IngestionPipeline( new IngestionPipelineSettings { BatchSize = MaxBatchSize, - MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.ErrorIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.ErrorIngestionMaxParallelWriters), logger), + MaxWriters = writers, BatchTimeout = settings.ErrorIngestionBatchTimeout }, IngestBatch,