From 7bf4fcf6ceaa14ab9ba25dc579c5902b8904b5d1 Mon Sep 17 00:00:00 2001 From: John Simons Date: Fri, 21 Aug 2026 12:12:34 +1000 Subject: [PATCH] Report error ingestion through OpenTelemetry The audit instance has had OpenTelemetry ingestion metrics for a while: batch and per message durations as histograms, a failure counter, and a gauge for consecutive batch failures, all exportable over OTLP and graphable. The error instance had counters that a hosted service printed to the log every five seconds when PrintMetrics was on, which cannot be graphed and cannot be alerted on. The error instance now reports the same shape under sc.error.ingestion.* on the meter Particular.ServiceControl, with OtlpEndpointUrl wired up the same way it is on the audit instance. Instrument names are what dashboards are built on, so both instances now have a test that pins theirs, and the audit names are unchanged. Where the two genuinely differ: - The error side keeps the bulk insert timing it already had, as sc.error.ingestion.storage_duration_seconds. It is the part of a batch that is neither announcing nor forwarding, and the audit ingestor has no equivalent. - message.category splits retry-confirmation from failed-message, which is the same split ErrorIngestor makes when it divides a batch, and the two cost very different amounts of work. The scopes behind all of this move into ServiceControl.Infrastructure: one batch, one message, one failure, one plain duration, and the histogram buckets they share so both instances stay comparable on the same dashboard. What stays per instance is what actually differs, which is instrument names, descriptions and the tags. Hosting audit ingestion in the primary will pick these up rather than copy them. The legacy counters leave the ingestion path, and with them the last thing that called Metrics.GetCounter or GetMeter anywhere. The reporter and PrintMetrics stay registered and now have nothing to report, so removing that plumbing, and the setting with it, is a decision on its own rather than a side effect of this. The consecutive batch failure gauge is updated with Interlocked and read with Volatile.Read. It was a plain field, which was safe while one writer owned the loop and loses increments now that several batches can fail at once, on the one number an alert would be built on. --- .../Auditing/Metrics/IngestionMetricsTests.cs | 98 ++++++++++++++++++ .../Auditing/AuditIngestionFaultPolicy.cs | 4 +- .../Auditing/Metrics/BatchMetrics.cs | 33 ------- .../Auditing/Metrics/ErrorMetrics.cs | 21 ---- .../Auditing/Metrics/IngestionMetrics.cs | 14 +-- .../Metrics/IngestionMetricsConfiguration.cs | 6 +- .../Auditing/Metrics/MessageMetrics.cs | 25 ----- .../Ingestion/Metrics/BatchMetrics.cs | 30 ++++++ .../Ingestion/Metrics/DurationScope.cs | 15 +++ .../Ingestion/Metrics/FailureMetrics.cs | 24 +++++ .../Ingestion/Metrics/IngestionDurations.cs | 12 +++ .../Ingestion/Metrics/MessageMetrics.cs | 27 +++++ ...rovals.PlatformSampleSettings.approved.txt | 1 + .../Metrics/IngestionMetricsTests.cs | 99 +++++++++++++++++++ .../HostApplicationBuilderExtensions.cs | 44 ++++++++- .../Infrastructure/Settings/Settings.cs | 2 + .../Operations/ErrorIngestion.cs | 39 ++++---- .../Operations/ErrorIngestionFaultPolicy.cs | 8 +- .../Operations/ErrorIngestor.cs | 15 ++- .../Operations/ErrorProcessor.cs | 5 - .../Operations/Metrics/IngestionMetrics.cs | 79 +++++++++++++++ .../Metrics/IngestionMetricsConfiguration.cs | 26 +++++ src/ServiceControl/ServiceControl.csproj | 3 + 23 files changed, 503 insertions(+), 127 deletions(-) create mode 100644 src/ServiceControl.Audit.UnitTests/Auditing/Metrics/IngestionMetricsTests.cs delete mode 100644 src/ServiceControl.Audit/Auditing/Metrics/BatchMetrics.cs delete mode 100644 src/ServiceControl.Audit/Auditing/Metrics/ErrorMetrics.cs delete mode 100644 src/ServiceControl.Audit/Auditing/Metrics/MessageMetrics.cs create mode 100644 src/ServiceControl.Infrastructure/Ingestion/Metrics/BatchMetrics.cs create mode 100644 src/ServiceControl.Infrastructure/Ingestion/Metrics/DurationScope.cs create mode 100644 src/ServiceControl.Infrastructure/Ingestion/Metrics/FailureMetrics.cs create mode 100644 src/ServiceControl.Infrastructure/Ingestion/Metrics/IngestionDurations.cs create mode 100644 src/ServiceControl.Infrastructure/Ingestion/Metrics/MessageMetrics.cs create mode 100644 src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs create mode 100644 src/ServiceControl/Operations/Metrics/IngestionMetrics.cs create mode 100644 src/ServiceControl/Operations/Metrics/IngestionMetricsConfiguration.cs diff --git a/src/ServiceControl.Audit.UnitTests/Auditing/Metrics/IngestionMetricsTests.cs b/src/ServiceControl.Audit.UnitTests/Auditing/Metrics/IngestionMetricsTests.cs new file mode 100644 index 0000000000..1357d7705b --- /dev/null +++ b/src/ServiceControl.Audit.UnitTests/Auditing/Metrics/IngestionMetricsTests.cs @@ -0,0 +1,98 @@ +namespace ServiceControl.Audit.UnitTests.Auditing.Metrics; + +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Audit.Auditing.Metrics; + +/// +/// Instrument names are what dashboards and alerts are built on, so they are a published contract +/// and not an implementation detail. +/// +[TestFixture] +class IngestionMetricsTests +{ + [SetUp] + public void CreateMeterFactory() => provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + + [TearDown] + public void DisposeMeterFactory() => provider.Dispose(); + + [Test] + public void The_meter_publishes_the_instruments_it_is_named_for() + { + var published = new List(); + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, _) => + { + if (BelongsToThisTest(instrument)) + { + published.Add(instrument.Name); + } + } + }; + + listener.Start(); + + _ = new IngestionMetrics(MeterFactory); + + Assert.That(published.Order(), Is.EqualTo(new[] + { + "sc.audit.ingestion.batch_duration_seconds", + "sc.audit.ingestion.consecutive_batch_failures_total", + "sc.audit.ingestion.failures_total", + "sc.audit.ingestion.message_duration_seconds" + })); + } + + [Test] + public void Concurrent_batch_failures_are_all_counted() + { + var metrics = new IngestionMetrics(MeterFactory); + + const int failedBatches = 1000; + + Parallel.For(0, failedBatches, _ => + { + using var batch = metrics.BeginBatch(maxBatchSize: 1); + }); + + Assert.That(ReadConsecutiveBatchFailures(), Is.EqualTo(failedBatches)); + } + + long ReadConsecutiveBatchFailures() + { + long value = -1; + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, activeListener) => + { + if (BelongsToThisTest(instrument) && instrument.Name == "sc.audit.ingestion.consecutive_batch_failures_total") + { + activeListener.EnableMeasurementEvents(instrument); + } + } + }; + + listener.SetMeasurementEventCallback((_, measurement, _, _) => value = measurement); + listener.Start(); + listener.RecordObservableInstruments(); + + return value; + } + + // Every fixture in the run shares the meter name, so the factory is what tells the instruments + // created here apart from the ones another test left behind. + bool BelongsToThisTest(Instrument instrument) => + instrument.Meter.Name == IngestionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory); + + IMeterFactory MeterFactory => provider.GetRequiredService(); + + ServiceProvider provider; +} diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs b/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs index c87ca5d5dc..b9d323747f 100644 --- a/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs +++ b/src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs @@ -38,12 +38,12 @@ public AuditIngestionFaultPolicy( public async Task OnError(ErrorContext errorContext, CancellationToken cancellationToken = default) { - using var errorMetrics = metrics.BeginErrorHandling(errorContext); + using var failureMetrics = metrics.BeginErrorHandling(errorContext); //Same as recoverability policy in NServiceBusFactory if (errorContext.ImmediateProcessingFailures < 3) { - errorMetrics.Retry(); + failureMetrics.Retry(); return ErrorHandleResult.RetryRequired; } diff --git a/src/ServiceControl.Audit/Auditing/Metrics/BatchMetrics.cs b/src/ServiceControl.Audit/Auditing/Metrics/BatchMetrics.cs deleted file mode 100644 index 41d2994fa9..0000000000 --- a/src/ServiceControl.Audit/Auditing/Metrics/BatchMetrics.cs +++ /dev/null @@ -1,33 +0,0 @@ -namespace ServiceControl.Audit.Auditing.Metrics; - -using System; -using System.Diagnostics; -using System.Diagnostics.Metrics; - -public record BatchMetrics(int MaxBatchSize, Histogram BatchDuration, Action IsSuccess) : IDisposable -{ - public void Dispose() - { - var isSuccess = actualBatchSize > 0; - - IsSuccess(isSuccess); - - string result; - - if (isSuccess) - { - result = actualBatchSize == MaxBatchSize ? "full" : "partial"; - } - else - { - result = "failed"; - } - - BatchDuration.Record(sw.Elapsed.TotalSeconds, new TagList { { "result", result } }); - } - - public void Complete(int size) => actualBatchSize = size; - - int actualBatchSize = -1; - readonly Stopwatch sw = Stopwatch.StartNew(); -} \ No newline at end of file diff --git a/src/ServiceControl.Audit/Auditing/Metrics/ErrorMetrics.cs b/src/ServiceControl.Audit/Auditing/Metrics/ErrorMetrics.cs deleted file mode 100644 index 6e5200b63d..0000000000 --- a/src/ServiceControl.Audit/Auditing/Metrics/ErrorMetrics.cs +++ /dev/null @@ -1,21 +0,0 @@ -namespace ServiceControl.Audit.Auditing.Metrics; - -using System; -using System.Diagnostics.Metrics; -using NServiceBus.Transport; - -public record ErrorMetrics(ErrorContext Context, Counter Failures) : IDisposable -{ - public void Dispose() - { - var tags = IngestionMetrics.GetMessageTags(Context.Headers); - - tags.Add("result", retry ? "retry" : "stored-poison"); - - Failures.Add(1, tags); - } - - public void Retry() => retry = true; - - bool retry; -} \ No newline at end of file diff --git a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs index f8bd2d763a..d79a7e76f0 100644 --- a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs +++ b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs @@ -3,9 +3,11 @@ namespace ServiceControl.Audit.Auditing.Metrics; using System.Collections.Generic; using System.Diagnostics; using System.Diagnostics.Metrics; +using System.Threading; using EndpointPlugin.Messages.SagaState; using NServiceBus; using NServiceBus.Transport; +using ServiceControl.Infrastructure.Ingestion.Metrics; public class IngestionMetrics { @@ -20,17 +22,17 @@ public IngestionMetrics(IMeterFactory meterFactory) batchDuration = meter.CreateHistogram(BatchDurationInstrumentName, unit: "seconds", "Message batch processing duration in seconds"); ingestionDuration = meter.CreateHistogram(MessageDurationInstrumentName, unit: "seconds", description: "Audit message processing duration in seconds"); - consecutiveBatchFailureGauge = meter.CreateObservableGauge($"{InstrumentPrefix}.consecutive_batch_failures_total", () => consecutiveBatchFailures, description: "Consecutive audit ingestion batch failures"); + consecutiveBatchFailureGauge = meter.CreateObservableGauge($"{InstrumentPrefix}.consecutive_batch_failures_total", () => Volatile.Read(ref consecutiveBatchFailures), description: "Consecutive audit ingestion batch failures"); failureCounter = meter.CreateCounter($"{InstrumentPrefix}.failures_total", description: "Audit ingestion failure count"); } - public MessageMetrics BeginIngestion(MessageContext messageContext) => new(messageContext, ingestionDuration); + public MessageMetrics BeginIngestion(MessageContext messageContext) => new(GetMessageTags(messageContext.Headers), ingestionDuration); - public ErrorMetrics BeginErrorHandling(ErrorContext errorContext) => new(errorContext, failureCounter); + public FailureMetrics BeginErrorHandling(ErrorContext errorContext) => new(GetMessageTags(errorContext.Headers), failureCounter); public BatchMetrics BeginBatch(int maxBatchSize) => new(maxBatchSize, batchDuration, RecordBatchOutcome); - public static TagList GetMessageTags(Dictionary headers) + static TagList GetMessageTags(Dictionary headers) { var tags = new TagList(); @@ -50,11 +52,11 @@ void RecordBatchOutcome(bool success) { if (success) { - consecutiveBatchFailures = 0; + Interlocked.Exchange(ref consecutiveBatchFailures, 0); } else { - consecutiveBatchFailures++; + Interlocked.Increment(ref consecutiveBatchFailures); } } diff --git a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetricsConfiguration.cs b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetricsConfiguration.cs index b78b6cd62a..e153daf8af 100644 --- a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetricsConfiguration.cs +++ b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetricsConfiguration.cs @@ -1,6 +1,7 @@ namespace ServiceControl.Audit.Auditing.Metrics; using OpenTelemetry.Metrics; +using ServiceControl.Infrastructure.Ingestion.Metrics; public static class IngestionMetricsConfiguration { @@ -8,12 +9,11 @@ public static void AddIngestionMetrics(this MeterProviderBuilder builder) { builder.AddMeter(IngestionMetrics.MeterName); - // Note: Views can be replaced by new InstrumentAdvice { HistogramBucketBoundaries = [...] }; once we can update to the latest OpenTelemetry packages builder.AddView( instrumentName: IngestionMetrics.MessageDurationInstrumentName, - new ExplicitBucketHistogramConfiguration { Boundaries = [0.01, 0.05, 0.1, 0.5, 1, 5] }); + new ExplicitBucketHistogramConfiguration { Boundaries = IngestionDurations.BucketBoundaries }); builder.AddView( instrumentName: IngestionMetrics.BatchDurationInstrumentName, - new ExplicitBucketHistogramConfiguration { Boundaries = [0.01, 0.05, 0.1, 0.5, 1, 5] }); + new ExplicitBucketHistogramConfiguration { Boundaries = IngestionDurations.BucketBoundaries }); } } \ No newline at end of file diff --git a/src/ServiceControl.Audit/Auditing/Metrics/MessageMetrics.cs b/src/ServiceControl.Audit/Auditing/Metrics/MessageMetrics.cs deleted file mode 100644 index df85310365..0000000000 --- a/src/ServiceControl.Audit/Auditing/Metrics/MessageMetrics.cs +++ /dev/null @@ -1,25 +0,0 @@ -namespace ServiceControl.Audit.Auditing.Metrics; - -using System; -using System.Diagnostics; -using System.Diagnostics.Metrics; -using NServiceBus.Transport; - -public record MessageMetrics(MessageContext Context, Histogram Duration) : IDisposable -{ - public void Skipped() => result = "skipped"; - - public void Success() => result = "success"; - - public void Dispose() - { - var tags = IngestionMetrics.GetMessageTags(Context.Headers); - - tags.Add("result", result); - Duration.Record(sw.Elapsed.TotalSeconds, tags); - } - - string result = "failed"; - - readonly Stopwatch sw = Stopwatch.StartNew(); -} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Ingestion/Metrics/BatchMetrics.cs b/src/ServiceControl.Infrastructure/Ingestion/Metrics/BatchMetrics.cs new file mode 100644 index 0000000000..4ef08a1ed8 --- /dev/null +++ b/src/ServiceControl.Infrastructure/Ingestion/Metrics/BatchMetrics.cs @@ -0,0 +1,30 @@ +namespace ServiceControl.Infrastructure.Ingestion.Metrics; + +using System; +using System.Diagnostics; +using System.Diagnostics.Metrics; + +/// +/// One batch write. Leaving the scope without calling is what records the +/// batch as failed, so nothing has to be told about the exception that ended it. +/// +public sealed class BatchMetrics(int maxBatchSize, Histogram batchDuration, Action recordOutcome) : IDisposable +{ + public void Complete(int batchSize) => completedSize = batchSize; + + public void Dispose() + { + var succeeded = completedSize > 0; + + recordOutcome(succeeded); + + var result = succeeded + ? completedSize == maxBatchSize ? "full" : "partial" + : "failed"; + + batchDuration.Record(stopwatch.Elapsed.TotalSeconds, new TagList { { "result", result } }); + } + + int completedSize = -1; + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Ingestion/Metrics/DurationScope.cs b/src/ServiceControl.Infrastructure/Ingestion/Metrics/DurationScope.cs new file mode 100644 index 0000000000..16535614ac --- /dev/null +++ b/src/ServiceControl.Infrastructure/Ingestion/Metrics/DurationScope.cs @@ -0,0 +1,15 @@ +namespace ServiceControl.Infrastructure.Ingestion.Metrics; + +using System; +using System.Diagnostics; +using System.Diagnostics.Metrics; + +/// +/// How long the scope was open, in seconds, and nothing else. +/// +public sealed class DurationScope(Histogram duration) : IDisposable +{ + public void Dispose() => duration.Record(stopwatch.Elapsed.TotalSeconds); + + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Ingestion/Metrics/FailureMetrics.cs b/src/ServiceControl.Infrastructure/Ingestion/Metrics/FailureMetrics.cs new file mode 100644 index 0000000000..9b0f8abb2d --- /dev/null +++ b/src/ServiceControl.Infrastructure/Ingestion/Metrics/FailureMetrics.cs @@ -0,0 +1,24 @@ +namespace ServiceControl.Infrastructure.Ingestion.Metrics; + +using System; +using System.Diagnostics; +using System.Diagnostics.Metrics; + +/// +/// One message the ingestion could not handle. Leaving the scope without saying otherwise records +/// it as having been given up on and stored as a failed import. +/// +public sealed class FailureMetrics(TagList messageTags, Counter failures) : IDisposable +{ + public void Retry() => retry = true; + + public void Dispose() + { + var tags = messageTags; + tags.Add("result", retry ? "retry" : "stored-poison"); + + failures.Add(1, tags); + } + + bool retry; +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Ingestion/Metrics/IngestionDurations.cs b/src/ServiceControl.Infrastructure/Ingestion/Metrics/IngestionDurations.cs new file mode 100644 index 0000000000..155717eb24 --- /dev/null +++ b/src/ServiceControl.Infrastructure/Ingestion/Metrics/IngestionDurations.cs @@ -0,0 +1,12 @@ +namespace ServiceControl.Infrastructure.Ingestion.Metrics; + +/// +/// The histogram buckets every ingestion duration is reported in, shared so the instances stay +/// comparable on one dashboard. +/// +public static class IngestionDurations +{ + // Views can give way to new InstrumentAdvice { HistogramBucketBoundaries = ... } once we + // can update to the latest OpenTelemetry packages + public static readonly double[] BucketBoundaries = [0.01, 0.05, 0.1, 0.5, 1, 5]; +} \ No newline at end of file diff --git a/src/ServiceControl.Infrastructure/Ingestion/Metrics/MessageMetrics.cs b/src/ServiceControl.Infrastructure/Ingestion/Metrics/MessageMetrics.cs new file mode 100644 index 0000000000..ed444e0121 --- /dev/null +++ b/src/ServiceControl.Infrastructure/Ingestion/Metrics/MessageMetrics.cs @@ -0,0 +1,27 @@ +namespace ServiceControl.Infrastructure.Ingestion.Metrics; + +using System; +using System.Diagnostics; +using System.Diagnostics.Metrics; + +/// +/// One message, from being received to its batch being written. Leaving the scope without saying +/// otherwise records it as failed. +/// +public sealed class MessageMetrics(TagList messageTags, Histogram duration) : IDisposable +{ + public void Skipped() => result = "skipped"; + + public void Success() => result = "success"; + + public void Dispose() + { + var tags = messageTags; + tags.Add("result", result); + + duration.Record(stopwatch.Elapsed.TotalSeconds, tags); + } + + string result = "failed"; + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} \ 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 2d2f5fad5a..a638e3df4f 100644 --- a/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt +++ b/src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt @@ -59,6 +59,7 @@ "Port": 8888, "PersisterSpecificSettings": null, "PrintMetrics": false, + "OtlpEndpointUrl": null, "Hostname": "localhost", "VirtualDirectory": "", "HeartbeatGracePeriod": "00:00:40", diff --git a/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs b/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs new file mode 100644 index 0000000000..6da39ae9a8 --- /dev/null +++ b/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs @@ -0,0 +1,99 @@ +namespace ServiceControl.UnitTests.Operations.Metrics; + +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Linq; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Operations.Metrics; + +/// +/// Instrument names are what dashboards and alerts are built on, so they are a published contract +/// and not an implementation detail. +/// +[TestFixture] +class IngestionMetricsTests +{ + [SetUp] + public void CreateMeterFactory() => provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + + [TearDown] + public void DisposeMeterFactory() => provider.Dispose(); + + [Test] + public void The_meter_publishes_the_instruments_it_is_named_for() + { + var published = new List(); + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, _) => + { + if (BelongsToThisTest(instrument)) + { + published.Add(instrument.Name); + } + } + }; + + listener.Start(); + + _ = new IngestionMetrics(MeterFactory); + + Assert.That(published.Order(), Is.EqualTo(new[] + { + "sc.error.ingestion.batch_duration_seconds", + "sc.error.ingestion.consecutive_batch_failures_total", + "sc.error.ingestion.failures_total", + "sc.error.ingestion.message_duration_seconds", + "sc.error.ingestion.storage_duration_seconds" + })); + } + + [Test] + public void Concurrent_batch_failures_are_all_counted() + { + var metrics = new IngestionMetrics(MeterFactory); + + const int failedBatches = 1000; + + Parallel.For(0, failedBatches, _ => + { + using var batch = metrics.BeginBatch(maxBatchSize: 1); + }); + + Assert.That(ReadConsecutiveBatchFailures(), Is.EqualTo(failedBatches)); + } + + long ReadConsecutiveBatchFailures() + { + long value = -1; + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, activeListener) => + { + if (BelongsToThisTest(instrument) && instrument.Name == "sc.error.ingestion.consecutive_batch_failures_total") + { + activeListener.EnableMeasurementEvents(instrument); + } + } + }; + + listener.SetMeasurementEventCallback((_, measurement, _, _) => value = measurement); + listener.Start(); + listener.RecordObservableInstruments(); + + return value; + } + + // Every fixture in the run shares the meter name, so the factory is what tells the instruments + // created here apart from the ones another test left behind. + bool BelongsToThisTest(Instrument instrument) => + instrument.Meter.Name == IngestionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory); + + IMeterFactory MeterFactory => provider.GetRequiredService(); + + ServiceProvider provider; +} diff --git a/src/ServiceControl/HostApplicationBuilderExtensions.cs b/src/ServiceControl/HostApplicationBuilderExtensions.cs index d8790c9151..cbef15f93b 100644 --- a/src/ServiceControl/HostApplicationBuilderExtensions.cs +++ b/src/ServiceControl/HostApplicationBuilderExtensions.cs @@ -1,4 +1,4 @@ -namespace Particular.ServiceControl +namespace Particular.ServiceControl { using System; using System.Diagnostics; @@ -14,6 +14,7 @@ namespace Particular.ServiceControl using global::ServiceControl.Infrastructure.Metrics; using global::ServiceControl.Infrastructure.WebApi; using global::ServiceControl.Notifications.Email; + using global::ServiceControl.Operations.Metrics; using global::ServiceControl.Persistence; using global::ServiceControl.Transports; using Licensing; @@ -26,6 +27,8 @@ namespace Particular.ServiceControl using NServiceBus.Configuration.AdvancedExtensibility; using NServiceBus.Hosting; using NServiceBus.Transport; + using OpenTelemetry.Metrics; + using OpenTelemetry.Resources; using Particular.LicensingComponent; using ServiceBus.Management.Infrastructure; using ServiceBus.Management.Infrastructure.Installers; @@ -33,6 +36,8 @@ namespace Particular.ServiceControl static class HostApplicationBuilderExtensions { + static readonly string InstanceVersion = FileVersionInfo.GetVersionInfo(typeof(HostApplicationBuilderExtensions).Assembly.Location).ProductVersion; + public static void AddServiceControl(this IHostApplicationBuilder hostBuilder, Settings settings, EndpointConfiguration configuration, params ReadOnlySpan components) { if (!settings.ErrorIngestionOnly) @@ -93,6 +98,7 @@ public static void AddServiceControl(this IHostApplicationBuilder hostBuilder, S services.AddPersistence(settings); services.AddMetrics(settings.PrintMetrics); + hostBuilder.AddIngestionMetrics(settings); services.AddServiceControlHealthChecks(); if (settings.ErrorIngestionOnly) @@ -142,9 +148,43 @@ public static void AddServiceControlInstallers(this IHostApplicationBuilder host persistence.AddInstaller(hostApplicationBuilder.Services); } + public static void AddIngestionMetrics(this IHostApplicationBuilder hostBuilder, Settings settings) + { + hostBuilder.Services.AddSingleton(); + + if (string.IsNullOrEmpty(settings.OtlpEndpointUrl)) + { + return; + } + + if (!Uri.TryCreate(settings.OtlpEndpointUrl, UriKind.Absolute, out var otlpEndpoint)) + { + throw new UriFormatException($"Invalid OtlpEndpointUrl: {settings.OtlpEndpointUrl}"); + } + + hostBuilder.Services.AddOpenTelemetry() + .ConfigureResource(resource => resource.AddService( + serviceName: settings.InstanceName, + serviceVersion: InstanceVersion, + autoGenerateServiceInstanceId: true)) + .WithMetrics(metrics => + { + metrics.AddIngestionMetrics(); + metrics.AddOtlpExporter(exporter => exporter.Endpoint = otlpEndpoint); + + if (Debugger.IsAttached) + { + metrics.AddConsoleExporter(); + } + }); + + LoggerUtil.CreateStaticLogger(typeof(HostApplicationBuilderExtensions), settings.LoggingSettings.LogLevel) + .LogInformation("OpenTelemetry metrics exporter enabled: {OtlpEndpointUrl}", settings.OtlpEndpointUrl); + } + static void RecordStartup(Settings settings, EndpointConfiguration endpointConfiguration) { - var version = FileVersionInfo.GetVersionInfo(typeof(HostApplicationBuilderExtensions).Assembly.Location).ProductVersion; + var version = InstanceVersion; var startupMessage = $@" ------------------------------------------------------------- diff --git a/src/ServiceControl/Infrastructure/Settings/Settings.cs b/src/ServiceControl/Infrastructure/Settings/Settings.cs index 2c1232ec64..8be72ac55e 100644 --- a/src/ServiceControl/Infrastructure/Settings/Settings.cs +++ b/src/ServiceControl/Infrastructure/Settings/Settings.cs @@ -177,6 +177,8 @@ public string InstanceId public PersistenceSettings PersisterSpecificSettings { get; set; } public bool PrintMetrics => SettingsReader.Read(SettingsRootNamespace, "PrintMetrics"); + + public string OtlpEndpointUrl { get; set; } = SettingsReader.Read(SettingsRootNamespace, nameof(OtlpEndpointUrl)); public string Hostname { get; private set; } public string VirtualDirectory => SettingsReader.Read(SettingsRootNamespace, "VirtualDirectory", string.Empty); diff --git a/src/ServiceControl/Operations/ErrorIngestion.cs b/src/ServiceControl/Operations/ErrorIngestion.cs index 107ca9550c..d23623c6d4 100644 --- a/src/ServiceControl/Operations/ErrorIngestion.cs +++ b/src/ServiceControl/Operations/ErrorIngestion.cs @@ -2,11 +2,10 @@ { using System; using System.Collections.Generic; - using System.Diagnostics; using System.Threading; using System.Threading.Tasks; using Infrastructure; - using Infrastructure.Metrics; + using Metrics; using Microsoft.Extensions.Hosting; using Microsoft.Extensions.Logging; using NServiceBus; @@ -19,13 +18,11 @@ class ErrorIngestion : BackgroundService { - static readonly long FrequencyInMilliseconds = Stopwatch.Frequency / 1000; - public ErrorIngestion( Settings settings, ITransportCustomization transportCustomization, TransportSettings transportSettings, - Metrics metrics, + IngestionMetrics metrics, IFailedErrorImportDataStore dataStore, ErrorIngestionCustomCheck.State ingestionState, ErrorIngestor ingestor, @@ -40,27 +37,27 @@ public ErrorIngestion( this.ingestor = ingestor; this.unitOfWorkFactory = unitOfWorkFactory; this.applicationLifetime = applicationLifetime; + this.metrics = metrics; this.logger = logger; - receivedMeter = metrics.GetCounter("Error ingestion - received"); - batchSizeMeter = metrics.GetMeter("Error ingestion - batch size"); - batchDurationMeter = metrics.GetMeter("Error ingestion - batch processing duration", FrequencyInMilliseconds); if (!transportSettings.MaxConcurrency.HasValue) { throw new ArgumentException("MaxConcurrency is not set in TransportSettings"); } + MaxBatchSize = settings.ErrorIngestionBatchSize ?? transportSettings.MaxConcurrency.Value; + pipeline = new IngestionPipeline( new IngestionPipelineSettings { - BatchSize = settings.ErrorIngestionBatchSize ?? transportSettings.MaxConcurrency.Value, + BatchSize = MaxBatchSize, MaxWriters = IngestionSettingsReader.ResolveMaxParallelWriters(settings.ErrorIngestionMaxParallelWriters, unitOfWorkFactory.SupportsConcurrentBatches, nameof(settings.ErrorIngestionMaxParallelWriters), logger), BatchTimeout = settings.ErrorIngestionBatchTimeout }, IngestBatch, logger); - errorHandlingPolicy = new ErrorIngestionFaultPolicy(dataStore, settings.LoggingSettings, OnCriticalError, logger); + errorHandlingPolicy = new ErrorIngestionFaultPolicy(dataStore, settings.LoggingSettings, OnCriticalError, metrics, logger); watchdog = new Watchdog( "failed message ingestion", @@ -83,12 +80,12 @@ public override async Task StartAsync(CancellationToken cancellationToken = defa async Task IngestBatch(List contexts, CancellationToken cancellationToken) { - batchSizeMeter.Mark(contexts.Count); + // Leaving the scope without completing it is what records the batch as failed + using var batchMetrics = metrics.BeginBatch(MaxBatchSize); - using (batchDurationMeter.Measure()) - { - await ingestor.Ingest(contexts, messageDispatcher, cancellationToken); - } + await ingestor.Ingest(contexts, messageDispatcher, cancellationToken); + + batchMetrics.Complete(contexts.Count); } public override async Task StopAsync(CancellationToken cancellationToken = default) @@ -236,8 +233,11 @@ async Task StopAndTeardownInfrastructure(CancellationToken cancellationToken) async Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken) { + using var messageIngestionMetrics = metrics.BeginIngestion(messageContext); + if (settings.MessageFilter != null && settings.MessageFilter(messageContext)) { + messageIngestionMetrics.Skipped(); return; } @@ -249,10 +249,10 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati // Not much shutdown speed to gain but this will ensure endpoint.Stop will return. await using var cancellationTokenRegistration = cancellationToken.Register(() => _ = taskCompletionSource.TrySetCanceled()); - receivedMeter.Mark(); - await pipeline.Enqueue(messageContext, cancellationToken); await taskCompletionSource.Task; + + messageIngestionMetrics.Success(); } Task OnCriticalError(string failure, Exception exception, CancellationToken cancellationToken) @@ -318,11 +318,10 @@ async Task StopReceiving(CancellationToken cancellationToken) readonly Settings settings; readonly ITransportCustomization transportCustomization; readonly TransportSettings transportSettings; + readonly int MaxBatchSize; readonly Watchdog watchdog; readonly IngestionPipeline pipeline; - readonly Meter batchDurationMeter; - readonly Meter batchSizeMeter; - readonly Counter receivedMeter; + readonly IngestionMetrics metrics; readonly ErrorIngestor ingestor; readonly IIngestionUnitOfWorkFactory unitOfWorkFactory; readonly IHostApplicationLifetime applicationLifetime; diff --git a/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs b/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs index 0e5fb7b1d0..d65bb3c2ed 100644 --- a/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs +++ b/src/ServiceControl/Operations/ErrorIngestionFaultPolicy.cs @@ -9,6 +9,7 @@ using System.Threading.Tasks; using Configuration; using Infrastructure; + using Metrics; using Microsoft.Extensions.Logging; using NServiceBus.Transport; using Persistence; @@ -21,9 +22,10 @@ class ErrorIngestionFaultPolicy ImportFailureCircuitBreaker failureCircuitBreaker; - public ErrorIngestionFaultPolicy(IFailedErrorImportDataStore store, LoggingSettings loggingSettings, Func onCriticalError, ILogger logger) + public ErrorIngestionFaultPolicy(IFailedErrorImportDataStore store, LoggingSettings loggingSettings, Func onCriticalError, IngestionMetrics metrics, ILogger logger) { this.store = store; + this.metrics = metrics; this.logger = logger; failureCircuitBreaker = new ImportFailureCircuitBreaker(onCriticalError); @@ -36,9 +38,12 @@ public ErrorIngestionFaultPolicy(IFailedErrorImportDataStore store, LoggingSetti public async Task OnError(ErrorContext errorContext, CancellationToken cancellationToken = default) { + using var failureMetrics = metrics.BeginErrorHandling(errorContext); + //Same as recoverability policy in NServiceBusFactory if (errorContext.ImmediateProcessingFailures < 3) { + failureMetrics.Retry(); return ErrorHandleResult.RetryRequired; } @@ -99,6 +104,7 @@ static void WriteToEventLog(string message) EventLog.WriteEntry(EventSourceCreator.SourceName, message, EventLogEntryType.Error); } + readonly IngestionMetrics metrics; readonly ILogger logger; } } \ No newline at end of file diff --git a/src/ServiceControl/Operations/ErrorIngestor.cs b/src/ServiceControl/Operations/ErrorIngestor.cs index b548ec156d..b62917904d 100644 --- a/src/ServiceControl/Operations/ErrorIngestor.cs +++ b/src/ServiceControl/Operations/ErrorIngestor.cs @@ -8,7 +8,7 @@ using System.Threading.Tasks; using Contracts.Operations; using Infrastructure.DomainEvents; - using Infrastructure.Metrics; + using Metrics; using Microsoft.Extensions.Logging; using NServiceBus.Routing; using NServiceBus.Transport; @@ -20,9 +20,7 @@ public class ErrorIngestor { - static readonly long FrequencyInMilliseconds = Stopwatch.Frequency / 1000; - - public ErrorIngestor(Metrics metrics, + public ErrorIngestor(IngestionMetrics metrics, IEnumerable errorEnrichers, IEnumerable failedMessageEnrichers, IDomainEvents domainEvents, @@ -33,9 +31,8 @@ public ErrorIngestor(Metrics metrics, { this.unitOfWorkFactory = unitOfWorkFactory; this.settings = settings; + this.metrics = metrics; this.logger = logger; - bulkInsertDurationMeter = metrics.GetMeter("Error ingestion - bulk insert duration", FrequencyInMilliseconds); - var ingestedMeter = metrics.GetCounter("Error ingestion - ingested"); var enrichers = new IEnrichImportedErrorMessages[] { @@ -45,7 +42,7 @@ public ErrorIngestor(Metrics metrics, }.Concat(errorEnrichers).ToArray(); - errorProcessor = new ErrorProcessor(enrichers, failedMessageEnrichers.ToArray(), domainEvents, ingestedMeter, logger); + errorProcessor = new ErrorProcessor(enrichers, failedMessageEnrichers.ToArray(), domainEvents, logger); retryConfirmationProcessor = new RetryConfirmationProcessor(domainEvents); logQueueAddress = new UnicastAddressTag(transportCustomization.ToTransportQualifiedQueueName(this.settings.ErrorLogQueue)); } @@ -123,7 +120,7 @@ async Task> PersistFailedMessages(List> Process(IReadOnlyList>()) { @@ -177,7 +173,6 @@ static void RecordKnownEndpoints(EndpointDetails observedEndpoint, Dictionary(BatchDurationInstrumentName, unit: "seconds", description: "Message batch processing duration in seconds"); + ingestionDuration = meter.CreateHistogram(MessageDurationInstrumentName, unit: "seconds", description: "Error message processing duration in seconds"); + storageDuration = meter.CreateHistogram(StorageDurationInstrumentName, unit: "seconds", description: "Error ingestion batch storage write duration in seconds"); + consecutiveBatchFailureGauge = meter.CreateObservableGauge($"{InstrumentPrefix}.consecutive_batch_failures_total", () => Volatile.Read(ref consecutiveBatchFailures), description: "Consecutive error ingestion batch failures"); + failureCounter = meter.CreateCounter($"{InstrumentPrefix}.failures_total", description: "Error ingestion failure count"); + } + + public MessageMetrics BeginIngestion(MessageContext messageContext) => new(GetMessageTags(messageContext.Headers), ingestionDuration); + + public FailureMetrics BeginErrorHandling(ErrorContext errorContext) => new(GetMessageTags(errorContext.Headers), failureCounter); + + public BatchMetrics BeginBatch(int maxBatchSize) => new(maxBatchSize, batchDuration, RecordBatchOutcome); + + /// + /// The storage write on its own, which is the part of a batch that is neither announcing nor + /// forwarding. + /// + public DurationScope MeasureStorageWrite() => new(storageDuration); + + // The same split the ingestor makes: a retry acknowledgement resolves a message, everything + // else records a failure, and they cost quite different amounts of work. + static TagList GetMessageTags(Dictionary headers) + { + return new TagList + { + { + "message.category", + headers.ContainsKey(RetryConfirmationProcessor.SuccessfulRetryHeader) ? "retry-confirmation" : "failed-message" + } + }; + } + + void RecordBatchOutcome(bool success) + { + if (success) + { + Interlocked.Exchange(ref consecutiveBatchFailures, 0); + } + else + { + Interlocked.Increment(ref consecutiveBatchFailures); + } + } + + long consecutiveBatchFailures; + + readonly Histogram batchDuration; +#pragma warning disable IDE0052 + // this can be changed to Gauge once we can use the latest version of System.Diagnostics.DiagnosticSource + readonly ObservableGauge consecutiveBatchFailureGauge; +#pragma warning restore IDE0052 + readonly Histogram ingestionDuration; + readonly Histogram storageDuration; + readonly Counter failureCounter; + + const string MeterVersion = "0.1.0"; + const string InstrumentPrefix = "sc.error.ingestion"; +} \ No newline at end of file diff --git a/src/ServiceControl/Operations/Metrics/IngestionMetricsConfiguration.cs b/src/ServiceControl/Operations/Metrics/IngestionMetricsConfiguration.cs new file mode 100644 index 0000000000..103115baac --- /dev/null +++ b/src/ServiceControl/Operations/Metrics/IngestionMetricsConfiguration.cs @@ -0,0 +1,26 @@ +namespace ServiceControl.Operations.Metrics; + +using OpenTelemetry.Metrics; +using ServiceControl.Infrastructure.Ingestion.Metrics; + +public static class IngestionMetricsConfiguration +{ + public static void AddIngestionMetrics(this MeterProviderBuilder builder) + { + builder.AddMeter(IngestionMetrics.MeterName); + + foreach (var instrumentName in DurationInstruments) + { + builder.AddView( + instrumentName, + new ExplicitBucketHistogramConfiguration { Boundaries = IngestionDurations.BucketBoundaries }); + } + } + + static readonly string[] DurationInstruments = + [ + IngestionMetrics.MessageDurationInstrumentName, + IngestionMetrics.BatchDurationInstrumentName, + IngestionMetrics.StorageDurationInstrumentName + ]; +} \ No newline at end of file diff --git a/src/ServiceControl/ServiceControl.csproj b/src/ServiceControl/ServiceControl.csproj index 201bf2478b..4401575e13 100644 --- a/src/ServiceControl/ServiceControl.csproj +++ b/src/ServiceControl/ServiceControl.csproj @@ -35,6 +35,9 @@ + + +