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 @@ + + +