Add retention sweep metrics for cycle duration, rows deleted, and consecutive failures

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";

src/ServiceControl.Infrastructure/ServiceControlMeters.cs

@@ -0,0 +1,12 @@
+namespace ServiceControl.Infrastructure;
+
+/// <summary>
+/// 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.
+/// </summary>
+public static class ServiceControlMeters
+{
+    public const string Error = "Particular.ServiceControl";
+    public const string Audit = "Particular.ServiceControl.Audit";
+}

src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs

@@ -35,6 +35,7 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste
 
         if (settings.RunRetentionSweep)
         {
+            services.AddSingleton<RetentionMetrics>();
             services.AddHostedService<RetentionSweeper>();
         }
 

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<double> { HistogramBucketBoundaries = [0.1, 0.5, 1, 5, 15, 60, 300, 900] });
+
+        rowsDeleted = meter.CreateCounter<long>(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<Measurement<long>> ObserveConsecutiveFailures()
+    {
+        for (var entity = 0; entity < consecutiveFailures.Length; entity++)
+        {
+            yield return new Measurement<long>(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<double> cycleDuration;
+    readonly Counter<long> rowsDeleted;
+#pragma warning disable IDE0052
+    readonly ObservableGauge<long> 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";
+}
+
+/// <summary>
+/// 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.
+/// </summary>
+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();
+}

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<ServiceControlDbContext>();
 
-        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)

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+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<double>((instrument, measurement, tags, _) => Add(instrument, measurement, tags));
+        listener.SetMeasurementEventCallback<long>((instrument, measurement, tags, _) => Add(instrument, measurement, tags));
+        listener.Start();
+    }
+
+    public IReadOnlyList<Recorded> Of(string instrumentName, RetentionEntity entity)
+    {
+        lock (measurements)
+        {
+            return
+            [
+                .. measurements.Where(measurement =>
+                    measurement.InstrumentName == instrumentName &&
+                    Equals(measurement.Tags["retention.entity"], EntityTag(entity)))
+            ];
+        }
+    }
+
+    public IReadOnlyList<Recorded> 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<KeyValuePair<string, object>> tags)
+    {
+        var copied = new Dictionary<string, object>();
+
+        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<Recorded> measurements = [];
+
+    public sealed record Recorded(string InstrumentName, double Value, Dictionary<string, object> Tags)
+    {
+        public object Result => Tags["result"];
+    }
+}