diff --git a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs index d79a7e76f0..038fefb7df 100644 --- a/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs +++ b/src/ServiceControl.Audit/Auditing/Metrics/IngestionMetrics.cs @@ -7,11 +7,12 @@ namespace ServiceControl.Audit.Auditing.Metrics; using EndpointPlugin.Messages.SagaState; using NServiceBus; using NServiceBus.Transport; +using ServiceControl.Infrastructure; using ServiceControl.Infrastructure.Ingestion.Metrics; public class IngestionMetrics { - public const string MeterName = "Particular.ServiceControl.Audit"; + public const string MeterName = ServiceControlMeters.Audit; public static readonly string BatchDurationInstrumentName = $"{InstrumentPrefix}.batch_duration_seconds"; public static readonly string MessageDurationInstrumentName = $"{InstrumentPrefix}.message_duration_seconds"; diff --git a/src/ServiceControl.Infrastructure/ServiceControlMeters.cs b/src/ServiceControl.Infrastructure/ServiceControlMeters.cs new file mode 100644 index 0000000000..aaac716f21 --- /dev/null +++ b/src/ServiceControl.Infrastructure/ServiceControlMeters.cs @@ -0,0 +1,12 @@ +namespace ServiceControl.Infrastructure; + +/// +/// The meters each instance publishes on. Shared because persisters publish onto the meter their +/// host has already registered with the exporter, and the two assemblies cannot reference each +/// other. +/// +public static class ServiceControlMeters +{ + public const string Error = "Particular.ServiceControl"; + public const string Audit = "Particular.ServiceControl.Audit"; +} diff --git a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs index 632bb0cbbb..a6a1e7da15 100644 --- a/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs +++ b/src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs @@ -35,6 +35,7 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste if (settings.RunRetentionSweep) { + services.AddSingleton(); services.AddHostedService(); } diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs new file mode 100644 index 0000000000..34e384de7b --- /dev/null +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionMetrics.cs @@ -0,0 +1,120 @@ +namespace ServiceControl.Persistence.EFCore.Infrastructure; + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using ServiceControl.Infrastructure; + +public enum RetentionEntity +{ + FailedMessages, + EventLog, + GroupComments +} + +public class RetentionMetrics +{ + public const string MeterName = ServiceControlMeters.Error; + + public static readonly string CycleDurationInstrumentName = $"{InstrumentPrefix}.cycle_duration_seconds"; + public static readonly string RowsDeletedInstrumentName = $"{InstrumentPrefix}.rows_deleted_total"; + public static readonly string ConsecutiveFailuresInstrumentName = $"{InstrumentPrefix}.consecutive_failures_total"; + + public RetentionMetrics(IMeterFactory meterFactory) + { + var meter = meterFactory.Create(MeterName, MeterVersion); + + cycleDuration = meter.CreateHistogram( + CycleDurationInstrumentName, + unit: "seconds", + description: "Retention sweep pass duration in seconds", + tags: null, + // A sweep pass is sub-second when it is keeping up and minutes long when it is working + // through a backlog, so the default boundaries resolve neither end. + advice: new InstrumentAdvice { HistogramBucketBoundaries = [0.1, 0.5, 1, 5, 15, 60, 300, 900] }); + + rowsDeleted = meter.CreateCounter(RowsDeletedInstrumentName, description: "Rows deleted by the retention sweep"); + consecutiveFailureGauge = meter.CreateObservableGauge(ConsecutiveFailuresInstrumentName, ObserveConsecutiveFailures, description: "Consecutive retention sweep failures"); + } + + public RetentionCycleMetrics BeginCycle(RetentionEntity entity, CancellationToken cancellationToken = default) => new(this, entity, cancellationToken); + + public void RecordRowsDeleted(RetentionEntity entity, int rows) => rowsDeleted.Add(rows, EntityTags[(int)entity]); + + internal void RecordCycle(RetentionEntity entity, TimeSpan elapsed, bool success) + { + var tags = EntityTags[(int)entity]; + tags.Add("result", success ? "success" : "failed"); + + cycleDuration.Record(elapsed.TotalSeconds, tags); + + if (success) + { + Interlocked.Exchange(ref consecutiveFailures[(int)entity], 0); + } + else + { + Interlocked.Increment(ref consecutiveFailures[(int)entity]); + } + } + + IEnumerable> ObserveConsecutiveFailures() + { + for (var entity = 0; entity < consecutiveFailures.Length; entity++) + { + yield return new Measurement(Volatile.Read(ref consecutiveFailures[entity]), EntityTags[entity]); + } + } + + static TagList EntityTag(string entity) => new() { { "retention.entity", entity } }; + + readonly long[] consecutiveFailures = new long[EntityTags.Length]; + + readonly Histogram cycleDuration; + readonly Counter rowsDeleted; +#pragma warning disable IDE0052 + readonly ObservableGauge consecutiveFailureGauge; +#pragma warning restore IDE0052 + + static readonly TagList[] EntityTags = + [ + EntityTag("failed_messages"), + EntityTag("event_log"), + EntityTag("group_comments") + ]; + + const string MeterVersion = "0.1.0"; + const string InstrumentPrefix = "sc.retention"; +} + +/// +/// One pass of the retention sweep. A pass interrupted by shutdown is not a measurement of +/// anything, so a cancelled cycle records neither a duration nor a failure. +/// +public sealed class RetentionCycleMetrics : IDisposable +{ + internal RetentionCycleMetrics(RetentionMetrics metrics, RetentionEntity entity, CancellationToken cancellationToken) + { + this.metrics = metrics; + this.entity = entity; + this.cancellationToken = cancellationToken; + } + + public void Complete() => completed = true; + + public void Dispose() + { + if (cancellationToken.IsCancellationRequested) + { + return; + } + + metrics.RecordCycle(entity, stopwatch.Elapsed, completed); + } + + bool completed; + + readonly RetentionMetrics metrics; + readonly RetentionEntity entity; + readonly CancellationToken cancellationToken; + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} diff --git a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs index 07811c4ad3..fd57ae2faa 100644 --- a/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs +++ b/src/ServiceControl.Persistence.EFCore/Infrastructure/RetentionSweeper.cs @@ -17,6 +17,7 @@ public class RetentionSweeper( TimeProvider timeProvider, IServiceScopeFactory serviceScopeFactory, IBodyStoragePersistence bodyStorage, + RetentionMetrics metrics, EFPersisterSettings settings) : BackgroundService { const int BatchSize = 1000; @@ -71,18 +72,25 @@ async Task Sweep(bool pace, CancellationToken cancellationToken) // group ids are deterministic, reattach a stale comment if the same failure ever recurs. async Task SweepOrphanedGroupComments(CancellationToken cancellationToken) { + using var cycle = metrics.BeginCycle(RetentionEntity.GroupComments, cancellationToken); using var scope = serviceScopeFactory.CreateScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); - await dbContext.GroupComments + var deleted = await dbContext.GroupComments .Where(comment => !dbContext.FailedMessageGroups.Any(group => group.GroupId == comment.GroupId)) .ExecuteDeleteAsync(cancellationToken); + + metrics.RecordRowsDeleted(RetentionEntity.GroupComments, deleted); + + cycle.Complete(); } // Event log items are insert-only and carry no external bodies, so each batch is a single // ordered DELETE. async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) { + using var cycle = metrics.BeginCycle(RetentionEntity.EventLog, cancellationToken); + var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.EventsRetentionPeriod; while (!cancellationToken.IsCancellationRequested) @@ -96,6 +104,8 @@ async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) .Take(BatchSize) .ExecuteDeleteAsync(cancellationToken); + metrics.RecordRowsDeleted(RetentionEntity.EventLog, deleted); + if (deleted < BatchSize) { break; @@ -106,10 +116,14 @@ async Task SweepEventLogItems(bool pace, CancellationToken cancellationToken) await Task.Delay(BatchPause, timeProvider, cancellationToken); } } + + cycle.Complete(); } async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) { + using var cycle = metrics.BeginCycle(RetentionEntity.FailedMessages, cancellationToken); + var cutoff = timeProvider.GetUtcNow().UtcDateTime - settings.ErrorRetentionPeriod; while (!cancellationToken.IsCancellationRequested) @@ -142,11 +156,13 @@ async Task SweepFailedMessages(bool pace, CancellationToken cancellationToken) // The predicate is re-asserted so a message that was re-failed (back to Unresolved) // between the select and the delete is left alone. The cascade removes its group rows. - await dbContext.FailedMessages + var deleted = await dbContext.FailedMessages .Where(failedMessage => ids.Contains(failedMessage.UniqueMessageId)) .Where(IsExpired(cutoff)) .ExecuteDeleteAsync(cancellationToken); + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, deleted); + if (expired.Count < BatchSize) { break; @@ -157,6 +173,8 @@ await dbContext.FailedMessages await Task.Delay(BatchPause, timeProvider, cancellationToken); } } + + cycle.Complete(); } async Task DeleteExternalBody(Guid uniqueMessageId, CancellationToken cancellationToken) diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs b/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs new file mode 100644 index 0000000000..687164d117 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/RecordedRetentionMetrics.cs @@ -0,0 +1,91 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Linq; +using ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Collects everything the retention instruments record, for the meter belonging to one factory. +/// Every fixture in the run shares the meter name, so the factory is what tells these instruments +/// apart from the ones another test left behind. +/// +sealed class RecordedRetentionMetrics : IDisposable +{ + public RecordedRetentionMetrics(IMeterFactory meterFactory) + { + listener = new MeterListener + { + InstrumentPublished = (instrument, activeListener) => + { + if (instrument.Meter.Name == RetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, meterFactory)) + { + activeListener.EnableMeasurementEvents(instrument); + } + } + }; + + listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => Add(instrument, measurement, tags)); + listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => Add(instrument, measurement, tags)); + listener.Start(); + } + + public IReadOnlyList Of(string instrumentName, RetentionEntity entity) + { + lock (measurements) + { + return + [ + .. measurements.Where(measurement => + measurement.InstrumentName == instrumentName && + Equals(measurement.Tags["retention.entity"], EntityTag(entity))) + ]; + } + } + + public IReadOnlyList Cycles(RetentionEntity entity) => Of(RetentionMetrics.CycleDurationInstrumentName, entity); + + public double RowsDeleted(RetentionEntity entity) => + Of(RetentionMetrics.RowsDeletedInstrumentName, entity).Sum(measurement => measurement.Value); + + public double ConsecutiveFailures(RetentionEntity entity) + { + listener.RecordObservableInstruments(); + + return Of(RetentionMetrics.ConsecutiveFailuresInstrumentName, entity)[^1].Value; + } + + public void Dispose() => listener.Dispose(); + + void Add(Instrument instrument, double value, ReadOnlySpan> tags) + { + var copied = new Dictionary(); + + foreach (var tag in tags) + { + copied[tag.Key] = tag.Value; + } + + lock (measurements) + { + measurements.Add(new Recorded(instrument.Name, value, copied)); + } + } + + static string EntityTag(RetentionEntity entity) => entity switch + { + RetentionEntity.FailedMessages => "failed_messages", + RetentionEntity.EventLog => "event_log", + RetentionEntity.GroupComments => "group_comments", + _ => throw new ArgumentOutOfRangeException(nameof(entity)) + }; + + readonly MeterListener listener; + readonly List measurements = []; + + public sealed record Recorded(string InstrumentName, double Value, Dictionary Tags) + { + public object Result => Tags["result"]; + } +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs new file mode 100644 index 0000000000..774f5a8a0a --- /dev/null +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionMetricsTests.cs @@ -0,0 +1,191 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Diagnostics.Metrics; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Persistence.EFCore.Infrastructure; + +/// +/// Instrument names are what dashboards and alerts are built on, so they are a published contract +/// and not an implementation detail. +/// +[TestFixture] +class RetentionMetricsTests +{ + [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 RetentionMetrics(MeterFactory); + + Assert.That(published.Order(), Is.EqualTo(new[] + { + "sc.retention.consecutive_failures_total", + "sc.retention.cycle_duration_seconds", + "sc.retention.rows_deleted_total" + })); + } + + [Test] + public void A_completed_cycle_is_recorded_as_a_success() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + using (var cycle = metrics.BeginCycle(RetentionEntity.EventLog)) + { + cycle.Complete(); + } + + var cycles = recorded.Cycles(RetentionEntity.EventLog); + + using (Assert.EnterMultipleScope()) + { + Assert.That(cycles, Has.Count.EqualTo(1)); + Assert.That(cycles[0].Result, Is.EqualTo("success")); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.Zero); + } + } + + [Test] + public void An_abandoned_cycle_is_recorded_as_a_failure() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(RetentionEntity.EventLog).Dispose(); + + var cycles = recorded.Cycles(RetentionEntity.EventLog); + + using (Assert.EnterMultipleScope()) + { + Assert.That(cycles, Has.Count.EqualTo(1)); + Assert.That(cycles[0].Result, Is.EqualTo("failed")); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.EqualTo(1)); + } + } + + [Test] + public void Consecutive_failures_are_counted_per_entity() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + metrics.BeginCycle(RetentionEntity.EventLog).Dispose(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.EqualTo(2)); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.EqualTo(1)); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.GroupComments), Is.Zero); + } + } + + [Test] + public void A_success_clears_the_failures_of_that_entity_alone() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose(); + metrics.BeginCycle(RetentionEntity.EventLog).Dispose(); + + using (var cycle = metrics.BeginCycle(RetentionEntity.FailedMessages)) + { + cycle.Complete(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.EventLog), Is.EqualTo(1)); + } + } + + [Test] + public void A_cycle_interrupted_by_shutdown_is_not_recorded() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + using (metrics.BeginCycle(RetentionEntity.FailedMessages, shutdown.Token)) + { + shutdown.Cancel(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages), Is.Empty); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); + } + } + + [Test] + public void Deleted_rows_are_counted_per_entity() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, 1000); + metrics.RecordRowsDeleted(RetentionEntity.FailedMessages, 7); + metrics.RecordRowsDeleted(RetentionEntity.GroupComments, 3); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.RowsDeleted(RetentionEntity.FailedMessages), Is.EqualTo(1007)); + Assert.That(recorded.RowsDeleted(RetentionEntity.GroupComments), Is.EqualTo(3)); + } + } + + [Test] + public void Concurrent_failures_are_all_counted() + { + var metrics = new RetentionMetrics(MeterFactory); + using var recorded = Listen(); + + const int failedCycles = 1000; + + Parallel.For(0, failedCycles, _ => metrics.BeginCycle(RetentionEntity.FailedMessages).Dispose()); + + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.EqualTo(failedCycles)); + } + + RecordedRetentionMetrics Listen() => new(MeterFactory); + + // 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 == RetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory); + + IMeterFactory MeterFactory => provider.GetRequiredService(); + + ServiceProvider provider; +} diff --git a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs index 122677d200..8948999fe6 100644 --- a/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs +++ b/src/ServiceControl.Persistence.Tests/EFCore/RetentionSweepTests.cs @@ -2,12 +2,15 @@ namespace ServiceControl.Persistence.Tests; using System; using System.Collections.Generic; +using System.Diagnostics.Metrics; using System.Linq; using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; using NUnit.Framework; using ServiceControl.EventLog; using ServiceControl.MessageFailures; using ServiceControl.Persistence.EFCore.Entities; +using ServiceControl.Persistence.EFCore.Infrastructure; using ServiceControl.Persistence.Infrastructure; class RetentionSweepTests : ErrorIngestionTestBase @@ -168,6 +171,48 @@ public async Task Archived_messages_are_swept_after_the_archiver_updates_the_tim Assert.That(await FindFailedMessage(messageId), Is.Null); } + [Test] + public async Task Counts_the_rows_it_deletes() + { + EFSettings.EventsRetentionPeriod = TimeSpan.FromDays(14); + + await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-31)); + await SeedFailedMessage(FailedMessageStatus.Resolved, Now.AddDays(-29)); + await Store(EventLogRow("expired", Now.AddDays(-15))); + + var expiredWithGroup = await SeedFailedMessage(FailedMessageStatus.Archived, Now.AddDays(-31)); + await GroupsStore.EditComment(await SeedGroup(expiredWithGroup), "Raised with the shipping team"); + + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.RowsDeleted(RetentionEntity.FailedMessages), Is.EqualTo(2), "the 29 day old message is still within retention"); + Assert.That(recorded.RowsDeleted(RetentionEntity.EventLog), Is.EqualTo(1)); + Assert.That(recorded.RowsDeleted(RetentionEntity.GroupComments), Is.EqualTo(1)); + } + } + + [Test] + public async Task Records_a_successful_cycle_for_every_pass() + { + using var recorded = ListenToRetentionMetrics(); + + await RunRetentionSweep(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.Cycles(RetentionEntity.FailedMessages).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.Cycles(RetentionEntity.EventLog).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.Cycles(RetentionEntity.GroupComments).Select(cycle => cycle.Result), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.ConsecutiveFailures(RetentionEntity.FailedMessages), Is.Zero); + } + } + + RecordedRetentionMetrics ListenToRetentionMetrics() => new(ServiceProvider.GetRequiredService()); + async Task SeedGroup(Guid uniqueMessageId) { var groupId = Guid.NewGuid().ToString(); diff --git a/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs b/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs index b877a21164..e395919797 100644 --- a/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs +++ b/src/ServiceControl/Operations/Metrics/IngestionMetrics.cs @@ -5,11 +5,12 @@ namespace ServiceControl.Operations.Metrics; using System.Diagnostics.Metrics; using System.Threading; using NServiceBus.Transport; +using ServiceControl.Infrastructure; using ServiceControl.Infrastructure.Ingestion.Metrics; public class IngestionMetrics { - public const string MeterName = "Particular.ServiceControl"; + public const string MeterName = ServiceControlMeters.Error; public static readonly string BatchDurationInstrumentName = $"{InstrumentPrefix}.batch_duration_seconds"; public static readonly string MessageDurationInstrumentName = $"{InstrumentPrefix}.message_duration_seconds";