Report audit retention metrics

docs/telemetry.md

@@ -107,33 +107,34 @@ Example Grafana dashboard: https://github.com/andreasohlund/Docker/blob/main/ote
 
 ## Retention
 
-Only the SQL persisters report retention. On RavenDB the expiry is metadata on each document and the deletes happen inside the Raven server, so there is no sweep of ours to measure.
+The SQL Server and PostgreSQL persisters of both the error and the audit instance report retention. On RavenDB the expiry is metadata on each document and the deletes happen inside the Raven server, so there is no sweep of ours to measure.
 
-Meter `Particular.ServiceControl`, the same meter the instance uses for its ingestion. The prefix carries no instance segment, so when the audit SQL persister reports its own retention it uses these same names, and `job` (or `exported_job`) separates the two.
+Each instance reports on the meter it uses for its ingestion: `Particular.ServiceControl` on the error instance and `Particular.ServiceControl.Audit` on the audit instance. The instrument names carry no instance segment, so both instances use the names below, and `job` (or `exported_job`) separates the two.
 
-The sweep runs hourly and makes one pass per kind of row. Each pass is measured on its own, tagged `retention.entity`: `failed_messages`, `event_log` or `group_comments`.
+The sweep runs hourly and makes one pass per kind of row. Each pass is measured on its own, tagged `retention.entity`: `failed_messages`, `event_log` or `group_comments` on the error instance, and `audit_messages` or `saga_snapshots` on the audit instance.
 
 - `sc.retention.cycle_duration_seconds` - Retention sweep pass duration in seconds
   - `retention.entity` - Which pass this was
   - `result` - The outcome: `success`, `failed`, or `cancelled` if shutdown cut the pass short
-- `sc.retention.rows_deleted_total` - Rows deleted by the retention sweep
+- `sc.retention.rows_deleted_total` - Rows deleted by the retention sweep. Error instance only. The audit instance leaves it out on both databases so they publish the same instruments, because its PostgreSQL sweep drops whole partitions and gets no row count back.
   - `retention.entity` - Which pass deleted them
 - `sc.retention.consecutive_failures_total` - Consecutive failures of that pass
   - `retention.entity` - Which pass is failing
 
 ### Reading the retention metrics
 
-- Rows reclaimed per hour: `sum(rate(sc_retention_rows_deleted_total[1h])) by (exported_job,retention_entity)`
+- Rows reclaimed per hour, error instance only: `sum(rate(sc_retention_rows_deleted_total[1h])) by (exported_job,retention_entity)`
 - Sweep duration: `histogram_quantile(0.9,sum(rate(sc_retention_cycle_duration_seconds_bucket[6h])) by (le,exported_job,retention_entity))`
 - Retention is broken: `max(sc_retention_consecutive_failures_total) by (exported_job,retention_entity) > 2`
 
 What the shapes mean:
 
 - `consecutive_failures_total` above zero is the signal that rows are no longer being reclaimed. Nothing else in the product reports this, and the database grows without bound while it lasts.
-- `rows_deleted_total` flat at zero across a long window is only healthy if the instance is also not ingesting. Deletion stopping while ingestion continues means the retention window is not being enforced.
+- On the error instance, `rows_deleted_total` flat at zero across a long window is only healthy if the instance is also not ingesting. Deletion stopping while ingestion continues means the retention window is not being enforced.
 - A cycle duration climbing towards the hourly interval means the sweep is no longer keeping up with the arrival rate, and each run starts further behind than the last.
 - A body store that refuses a delete fails the `failed_messages` pass rather than orphaning the body, so an expired credential or a changed permission shows as a climbing gauge instead of storage that quietly keeps growing. Those rows stay until a sweep can delete the body and the row together.
-- Each pass is isolated, so one kind of row failing to be reclaimed does not stop the others. Every pass reports a result on every run, and the gauge is per pass, so alert across all entities rather than on any one of them.
+- On the audit instance, a pass that leaves expired rows because its table is too busy to drop them is recorded as `failed`. The "Audit retention" custom check counts that sweep as failed too. A sweep skipped because another audit instance holds the retention lock records nothing.
+- Each pass is isolated, so one kind of row failing to be reclaimed does not stop the others. Every pass that runs reports a result, and the gauge is per pass, so alert across all entities rather than on any one of them.
 
 ## Monitoring
 

src/ServiceControl.Audit.Persistence.EFCore.PostgreSql/PostgreSqlAuditPartitionManager.cs

@@ -10,6 +10,7 @@ namespace ServiceControl.Audit.Persistence.EFCore.PostgreSql;
 using ServiceControl.Audit.Persistence.EFCore.DbContexts;
 using ServiceControl.Audit.Persistence.EFCore.Entities;
 using ServiceControl.Audit.Persistence.EFCore.Infrastructure;
+using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics;
 
 class PostgreSqlAuditPartitionManager(ILogger<PostgreSqlAuditPartitionManager> logger) : IAuditPartitionManager
 {
@@ -35,30 +36,20 @@ public async Task EnsurePartitions(AuditDbContext dbContext, DateTime from, Date
         }
     }
 
-    public async Task<IReadOnlyList<DateTime>> ListExpired(AuditDbContext dbContext, DateTime keepFrom, CancellationToken cancellationToken = default)
+    public async Task<IReadOnlyList<DateTime>> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default)
     {
-        var days = new SortedSet<DateTime>();
-        foreach (var table in Tables(dbContext))
-        {
-            foreach (var day in await PartitionDays(dbContext, table, cancellationToken))
-            {
-                if (day.AddDays(1) <= keepFrom)
-                {
-                    days.Add(day);
-                }
-            }
-        }
+        var days = await PartitionDays(dbContext, Table(dbContext, entity), cancellationToken);
 
-        return [.. days];
+        return [.. days.Where(day => day.AddDays(1) <= keepFrom).Order()];
     }
 
     // A partition holds a whole day, so the bodies are deleted by day rather than by hour.
     public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime day, CancellationToken cancellationToken = default) =>
         bodyStorage.DeleteBodiesForDay(day, cancellationToken);
 
-    public async Task<DropResult> DropExpired(AuditDbContext dbContext, DateTime day, int batchSize, CancellationToken cancellationToken = default)
+    public async Task<DropResult> DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime day, int batchSize, CancellationToken cancellationToken = default)
     {
-        var sql = string.Concat(Tables(dbContext).Select(table => $"DROP TABLE IF EXISTS {table.Partition(day)};\n"));
+        var sql = $"DROP TABLE IF EXISTS {Table(dbContext, entity).Partition(day)};";
 
         return await WithShortLockWait(dbContext, sql, cancellationToken)
             ? new DropResult(RowsDeleted: 0, Completed: true)
@@ -122,6 +113,13 @@ static Task<bool> WithShortLockWait(AuditDbContext dbContext, string sql, Cancel
 
     static AuditTable[] Tables(AuditDbContext dbContext) => [Table<AuditMessageEntity>(dbContext), Table<SagaSnapshotEntity>(dbContext)];
 
+    static AuditTable Table(AuditDbContext dbContext, AuditRetentionEntity entity) => entity switch
+    {
+        AuditRetentionEntity.AuditMessages => Table<AuditMessageEntity>(dbContext),
+        AuditRetentionEntity.SagaSnapshots => Table<SagaSnapshotEntity>(dbContext),
+        _ => throw new ArgumentOutOfRangeException(nameof(entity))
+    };
+
     static AuditTable Table<TEntity>(AuditDbContext dbContext) =>
         new(dbContext.Model.FindEntityType(typeof(TEntity))!.GetTableName()!, SchemaQualifiedTableName.For<TEntity>(dbContext), dbContext.Schema, dbContext.GetService<ISqlGenerationHelper>());
 

src/ServiceControl.Audit.Persistence.EFCore.SqlServer/SqlServerAuditPartitionManager.cs

@@ -3,44 +3,47 @@ namespace ServiceControl.Audit.Persistence.EFCore.SqlServer;
 using Microsoft.EntityFrameworkCore;
 using ServiceControl.Audit.Persistence.EFCore.DbContexts;
 using ServiceControl.Audit.Persistence.EFCore.Infrastructure;
+using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics;
 
 // No partitions, because a full text index cannot be aligned to a partition scheme.
 class SqlServerAuditPartitionManager : IAuditPartitionManager
 {
     public Task EnsurePartitions(AuditDbContext dbContext, DateTime fromHour, DateTime toHourExclusive, CancellationToken cancellationToken = default) =>
         Task.CompletedTask;
 
-    public async Task<IReadOnlyList<DateTime>> ListExpired(AuditDbContext dbContext, DateTime keepFrom, CancellationToken cancellationToken = default)
+    public async Task<IReadOnlyList<DateTime>> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default)
     {
-        var messageHours = dbContext.AuditMessages.Where(message => message.CreatedOn < keepFrom).Select(message => message.CreatedOn);
-        var snapshotHours = dbContext.SagaSnapshots.Where(snapshot => snapshot.CreatedOn < keepFrom).Select(snapshot => snapshot.CreatedOn);
+        var hours = entity switch
+        {
+            AuditRetentionEntity.AuditMessages => dbContext.AuditMessages.Where(message => message.CreatedOn < keepFrom).Select(message => message.CreatedOn),
+            AuditRetentionEntity.SagaSnapshots => dbContext.SagaSnapshots.Where(snapshot => snapshot.CreatedOn < keepFrom).Select(snapshot => snapshot.CreatedOn),
+            _ => throw new ArgumentOutOfRangeException(nameof(entity))
+        };
 
-        return await messageHours.Union(snapshotHours).OrderBy(hour => hour).ToListAsync(cancellationToken);
+        return await hours.Distinct().OrderBy(hour => hour).ToListAsync(cancellationToken);
     }
 
     public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime hour, CancellationToken cancellationToken = default) =>
         bodyStorage.DeleteBodiesForHour(hour, cancellationToken);
 
-    public async Task<DropResult> DropExpired(AuditDbContext dbContext, DateTime hour, int batchSize, CancellationToken cancellationToken = default)
+    public async Task<DropResult> DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime hour, int batchSize, CancellationToken cancellationToken = default)
     {
-        var messagesDeleted = await dbContext.AuditMessages
-            .Where(message => message.CreatedOn == hour)
-            .OrderBy(message => message.Id)
-            .Take(batchSize)
-            .ExecuteDeleteAsync(cancellationToken);
-
-        if (messagesDeleted == batchSize)
+        var deleted = entity switch
         {
-            return new DropResult(messagesDeleted, Completed: false);
-        }
-
-        var snapshotsDeleted = await dbContext.SagaSnapshots
-            .Where(snapshot => snapshot.CreatedOn == hour)
-            .OrderBy(snapshot => snapshot.Id)
-            .Take(batchSize)
-            .ExecuteDeleteAsync(cancellationToken);
-
-        return new DropResult(messagesDeleted + snapshotsDeleted, Completed: snapshotsDeleted < batchSize);
+            AuditRetentionEntity.AuditMessages => await dbContext.AuditMessages
+                .Where(message => message.CreatedOn == hour)
+                .OrderBy(message => message.Id)
+                .Take(batchSize)
+                .ExecuteDeleteAsync(cancellationToken),
+            AuditRetentionEntity.SagaSnapshots => await dbContext.SagaSnapshots
+                .Where(snapshot => snapshot.CreatedOn == hour)
+                .OrderBy(snapshot => snapshot.Id)
+                .Take(batchSize)
+                .ExecuteDeleteAsync(cancellationToken),
+            _ => throw new ArgumentOutOfRangeException(nameof(entity))
+        };
+
+        return new DropResult(deleted, Completed: deleted < batchSize);
     }
 
     public Task<DateTime?> ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) =>

src/ServiceControl.Audit.Persistence.EFCore/Abstractions/EFPersistenceBase.cs

@@ -7,6 +7,7 @@ namespace ServiceControl.Audit.Persistence.EFCore.Abstractions;
 using ServiceControl.Audit.Persistence.EFCore.Implementation.BodyStorage;
 using ServiceControl.Audit.Persistence.EFCore.Infrastructure;
 using ServiceControl.Audit.Persistence.EFCore.Implementation.UnitOfWork;
+using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics;
 using ServiceControl.Audit.Persistence.UnitOfWork;
 
 abstract class EFPersistenceBase(EFPersisterSettings settings) : IPersistence
@@ -26,6 +27,7 @@ public void AddPersistence(IServiceCollection services)
         AddEnvironmentProbes(services);
 
         services.AddSingleton<AuditRetentionCustomCheck.State>();
+        services.AddSingleton<AuditRetentionMetrics>();
         services.AddHostedService<AuditRetention>();
 
         if (services.SingleOrDefault(s => s.ServiceType == typeof(EndpointConfiguration)) is

src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditRetention.cs

@@ -6,13 +6,15 @@ namespace ServiceControl.Audit.Persistence.EFCore.Implementation;
 using ServiceControl.Audit.Persistence.EFCore.Abstractions;
 using ServiceControl.Audit.Persistence.EFCore.DbContexts;
 using ServiceControl.Audit.Persistence.EFCore.Infrastructure;
+using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics;
 
 sealed class AuditRetention(
     IServiceScopeFactory scopeFactory,
     IAuditPartitionManager partitions,
     IRetentionLock retentionLock,
     AuditRetentionCustomCheck.State retentionState,
     EFPersisterSettings settings,
+    AuditRetentionMetrics metrics,
     TimeProvider timeProvider,
     ILogger<AuditRetention> logger,
     IAuditBodyStoragePersistence? bodyStorage = null) : BackgroundService
@@ -68,9 +70,10 @@ protected override async Task ExecuteAsync(CancellationToken cancellationToken =
 
     async Task Sweep(bool pace, CancellationToken cancellationToken)
     {
+        var passesStarted = false;
         try
         {
-            DateTime? deferred;
+            List<string> passFailures = [];
             await using (var handle = await retentionLock.TryAcquire(cancellationToken))
             {
                 if (handle is null)
@@ -79,13 +82,23 @@ async Task Sweep(bool pace, CancellationToken cancellationToken)
                     return;
                 }
 
-                deferred = await RemoveExpired(pace, cancellationToken);
+                var now = timeProvider.GetUtcNow().UtcDateTime;
+                await EnsurePartitions(AuditHours.Truncate(now), cancellationToken);
+
+                var keepFrom = AuditHours.Truncate(now - settings.AuditRetentionPeriod);
+                passesStarted = true;
+                foreach (var entity in Entities)
+                {
+                    if (await RunPass(entity, keepFrom, pace, cancellationToken) is { } failure)
+                    {
+                        passFailures.Add(failure);
+                    }
+                }
             }
 
-            // A table that stays too busy to drop keeps every expired row, so it counts against the check like an error.
-            if (deferred is { } start)
+            if (passFailures.Count > 0)
             {
-                retentionState.SweepFailed($"The audit tables were too busy to drop the rows from {start:u}.");
+                retentionState.SweepFailed(string.Join("; ", passFailures));
             }
             else
             {
@@ -95,25 +108,62 @@ async Task Sweep(bool pace, CancellationToken cancellationToken)
 #pragma warning disable PS0019 // Filtered on the token alone because SqlClient reports a cancelled command as a SqlException.
         catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
         {
+            // No pass ran, so each one is recorded as failed and the gauge still climbs.
+            if (!passesStarted)
+            {
+                foreach (var entity in Entities)
+                {
+                    metrics.BeginCycle(entity, cancellationToken).Dispose();
+                }
+            }
+
             retentionState.SweepFailed(ex.Message);
             throw;
         }
 #pragma warning restore PS0019
     }
 
-    // Returns the start of the expired rows it had to leave because the tables were too busy, or null when it removed them all.
-    async Task<DateTime?> RemoveExpired(bool pace, CancellationToken cancellationToken)
+    async Task EnsurePartitions(DateTime hour, CancellationToken cancellationToken)
     {
         await using var scope = scopeFactory.CreateAsyncScope();
         var dbContext = scope.ServiceProvider.GetRequiredService<AuditDbContext>();
 
-        var now = timeProvider.GetUtcNow().UtcDateTime;
-        var hour = AuditHours.Truncate(now);
         await partitions.EnsurePartitions(dbContext, hour, hour + AuditHours.Lookahead, cancellationToken);
+    }
+
+    // Each pass is isolated so one failing kind of row does not stop the other from being reclaimed.
+    // Returns what went wrong for the custom check to report, or null when the pass succeeded.
+    async Task<string?> RunPass(AuditRetentionEntity entity, DateTime keepFrom, bool pace, CancellationToken cancellationToken)
+    {
+        using var cycle = metrics.BeginCycle(entity, cancellationToken);
+
+        try
+        {
+            // A busy table keeps every expired row from the hour where it stopped, so the pass fails rather than letting retention stall with no warning.
+            if (await RemoveExpired(entity, keepFrom, pace, cancellationToken) is { } deferred)
+            {
+                return $"{entity}: the table was too busy to drop the rows from {deferred:u}";
+            }
+
+            cycle.Complete();
+            return null;
+        }
+#pragma warning disable PS0019 // Filtered on the token alone because SqlClient reports a cancelled command as a SqlException.
+        catch (Exception ex) when (!cancellationToken.IsCancellationRequested)
+        {
+            logger.LogError(ex, "Error during the {RetentionEntity} audit retention pass", entity);
+            return $"{entity}: {ex.Message}";
+        }
+#pragma warning restore PS0019
+    }
 
-        var keepFrom = AuditHours.Truncate(now - settings.AuditRetentionPeriod);
+    // Returns the start of the expired rows where it stopped because the table was too busy, or null when it removed every expired row.
+    async Task<DateTime?> RemoveExpired(AuditRetentionEntity entity, DateTime keepFrom, bool pace, CancellationToken cancellationToken)
+    {
+        await using var scope = scopeFactory.CreateAsyncScope();
+        var dbContext = scope.ServiceProvider.GetRequiredService<AuditDbContext>();
 
-        foreach (var expired in await partitions.ListExpired(dbContext, keepFrom, cancellationToken))
+        foreach (var expired in await partitions.ListExpired(dbContext, entity, keepFrom, cancellationToken))
         {
             // Bodies go first: a failure here leaves the rows in place, so the next sweep retries the same hour.
             if (bodyStorage is not null)
@@ -124,11 +174,11 @@ async Task Sweep(bool pace, CancellationToken cancellationToken)
             DropResult drop;
             do
             {
-                drop = await partitions.DropExpired(dbContext, expired, BatchSize, cancellationToken);
+                drop = await partitions.DropExpired(dbContext, entity, expired, BatchSize, cancellationToken);
 
                 if (drop.Deferred)
                 {
-                    logger.LogInformation("Left the audit rows from {Start:u} for the next sweep because the audit tables were too busy to drop them", expired);
+                    logger.LogInformation("Left the {RetentionEntity} rows from {Start:u} for the next sweep because the table was too busy to drop them", entity, expired);
                     return expired;
                 }
 
@@ -138,9 +188,11 @@ async Task Sweep(bool pace, CancellationToken cancellationToken)
                 }
             } while (!drop.Completed);
 
-            logger.LogDebug("Removed the expired audit rows from {Start:u}", expired);
+            logger.LogDebug("Removed the expired {RetentionEntity} rows from {Start:u}", entity, expired);
         }
 
         return null;
     }
+
+    static readonly AuditRetentionEntity[] Entities = Enum.GetValues<AuditRetentionEntity>();
 }