diff --git a/docs/telemetry.md b/docs/telemetry.md index c9f6a98f33..68762a7c9e 100644 --- a/docs/telemetry.md +++ b/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 diff --git a/src/ServiceControl.Audit.Persistence.EFCore.PostgreSql/PostgreSqlAuditPartitionManager.cs b/src/ServiceControl.Audit.Persistence.EFCore.PostgreSql/PostgreSqlAuditPartitionManager.cs index da74858666..283a28f4e0 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore.PostgreSql/PostgreSqlAuditPartitionManager.cs +++ b/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 logger) : IAuditPartitionManager { @@ -35,30 +36,20 @@ public async Task EnsurePartitions(AuditDbContext dbContext, DateTime from, Date } } - public async Task> ListExpired(AuditDbContext dbContext, DateTime keepFrom, CancellationToken cancellationToken = default) + public async Task> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default) { - var days = new SortedSet(); - 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 DropExpired(AuditDbContext dbContext, DateTime day, int batchSize, CancellationToken cancellationToken = default) + public async Task 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 WithShortLockWait(AuditDbContext dbContext, string sql, Cancel static AuditTable[] Tables(AuditDbContext dbContext) => [Table(dbContext), Table(dbContext)]; + static AuditTable Table(AuditDbContext dbContext, AuditRetentionEntity entity) => entity switch + { + AuditRetentionEntity.AuditMessages => Table(dbContext), + AuditRetentionEntity.SagaSnapshots => Table(dbContext), + _ => throw new ArgumentOutOfRangeException(nameof(entity)) + }; + static AuditTable Table(AuditDbContext dbContext) => new(dbContext.Model.FindEntityType(typeof(TEntity))!.GetTableName()!, SchemaQualifiedTableName.For(dbContext), dbContext.Schema, dbContext.GetService()); diff --git a/src/ServiceControl.Audit.Persistence.EFCore.SqlServer/SqlServerAuditPartitionManager.cs b/src/ServiceControl.Audit.Persistence.EFCore.SqlServer/SqlServerAuditPartitionManager.cs index 2c4d19865e..584dffccdc 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore.SqlServer/SqlServerAuditPartitionManager.cs +++ b/src/ServiceControl.Audit.Persistence.EFCore.SqlServer/SqlServerAuditPartitionManager.cs @@ -3,6 +3,7 @@ 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 @@ -10,37 +11,39 @@ class SqlServerAuditPartitionManager : IAuditPartitionManager public Task EnsurePartitions(AuditDbContext dbContext, DateTime fromHour, DateTime toHourExclusive, CancellationToken cancellationToken = default) => Task.CompletedTask; - public async Task> ListExpired(AuditDbContext dbContext, DateTime keepFrom, CancellationToken cancellationToken = default) + public async Task> 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 DropExpired(AuditDbContext dbContext, DateTime hour, int batchSize, CancellationToken cancellationToken = default) + public async Task 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 ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) => diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Abstractions/EFPersistenceBase.cs b/src/ServiceControl.Audit.Persistence.EFCore/Abstractions/EFPersistenceBase.cs index 0799edd11d..830807d758 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore/Abstractions/EFPersistenceBase.cs +++ b/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(); + services.AddSingleton(); services.AddHostedService(); if (services.SingleOrDefault(s => s.ServiceType == typeof(EndpointConfiguration)) is diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditRetention.cs b/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditRetention.cs index ac76823776..466dbca45d 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditRetention.cs +++ b/src/ServiceControl.Audit.Persistence.EFCore/Implementation/AuditRetention.cs @@ -6,6 +6,7 @@ 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, @@ -13,6 +14,7 @@ sealed class AuditRetention( IRetentionLock retentionLock, AuditRetentionCustomCheck.State retentionState, EFPersisterSettings settings, + AuditRetentionMetrics metrics, TimeProvider timeProvider, ILogger 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 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 RemoveExpired(bool pace, CancellationToken cancellationToken) + async Task EnsurePartitions(DateTime hour, CancellationToken cancellationToken) { await using var scope = scopeFactory.CreateAsyncScope(); var dbContext = scope.ServiceProvider.GetRequiredService(); - 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 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 RemoveExpired(AuditRetentionEntity entity, DateTime keepFrom, bool pace, CancellationToken cancellationToken) + { + await using var scope = scopeFactory.CreateAsyncScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); - 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(); } diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/IAuditPartitionManager.cs b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/IAuditPartitionManager.cs index 1485477268..968aadc110 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/IAuditPartitionManager.cs +++ b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/IAuditPartitionManager.cs @@ -1,19 +1,20 @@ namespace ServiceControl.Audit.Persistence.EFCore.Infrastructure; using ServiceControl.Audit.Persistence.EFCore.DbContexts; +using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; interface IAuditPartitionManager { Task EnsurePartitions(AuditDbContext dbContext, DateTime from, DateTime toExclusive, CancellationToken cancellationToken = default); - Task> ListExpired(AuditDbContext dbContext, DateTime keepFrom, CancellationToken cancellationToken = default); + Task> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default); /// /// Deletes every body stored for the rows that removes for . /// Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime expired, CancellationToken cancellationToken = default); - Task DropExpired(AuditDbContext dbContext, DateTime start, int batchSize, CancellationToken cancellationToken = default); + Task DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime start, int batchSize, CancellationToken cancellationToken = default); Task ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default); } diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionCycleMetrics.cs b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionCycleMetrics.cs new file mode 100644 index 0000000000..bbd37efd9b --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionCycleMetrics.cs @@ -0,0 +1,40 @@ +namespace ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + +using System.Diagnostics; + +/// +/// Times one pass of the audit retention sweep and records it when disposed. +/// Call when the pass finishes, and it counts as a success even if shutdown has since been requested. +/// Disposed without , it counts as cancelled if shutdown was requested and as failed otherwise. +/// +public sealed class AuditRetentionCycleMetrics : IDisposable +{ + internal AuditRetentionCycleMetrics(AuditRetentionMetrics metrics, AuditRetentionEntity entity, CancellationToken cancellationToken) + { + this.metrics = metrics; + this.entity = entity; + this.cancellationToken = cancellationToken; + } + + // The pass is over once it completes, so the clock stops here rather than wherever the scope + // happens to be disposed. + public void Complete() + { + stopwatch.Stop(); + completed = true; + } + + public void Dispose() => metrics.RecordCycle(entity, stopwatch.Elapsed, Outcome); + + CycleOutcome Outcome => + completed ? CycleOutcome.Success + : cancellationToken.IsCancellationRequested ? CycleOutcome.Cancelled + : CycleOutcome.Failed; + + bool completed; + + readonly AuditRetentionMetrics metrics; + readonly AuditRetentionEntity entity; + readonly CancellationToken cancellationToken; + readonly Stopwatch stopwatch = Stopwatch.StartNew(); +} diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionEntity.cs b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionEntity.cs new file mode 100644 index 0000000000..fabff89383 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionEntity.cs @@ -0,0 +1,7 @@ +namespace ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + +public enum AuditRetentionEntity +{ + AuditMessages, + SagaSnapshots +} diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionMetrics.cs b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionMetrics.cs new file mode 100644 index 0000000000..819c6834ea --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionMetrics.cs @@ -0,0 +1,81 @@ +namespace ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + +using System.Diagnostics; +using System.Diagnostics.Metrics; +using ServiceControl.Infrastructure; + +public class AuditRetentionMetrics +{ + public const string MeterName = ServiceControlMeters.Audit; + + public static readonly string CycleDurationInstrumentName = $"{InstrumentPrefix}.cycle_duration_seconds"; + public static readonly string ConsecutiveFailuresInstrumentName = $"{InstrumentPrefix}.consecutive_failures_total"; + + public AuditRetentionMetrics(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] }); + + meter.CreateObservableGauge(ConsecutiveFailuresInstrumentName, ObserveConsecutiveFailures, description: "Consecutive retention sweep failures"); + } + + public AuditRetentionCycleMetrics BeginCycle(AuditRetentionEntity entity, CancellationToken cancellationToken = default) => new(this, entity, cancellationToken); + + internal void RecordCycle(AuditRetentionEntity entity, TimeSpan elapsed, CycleOutcome outcome) + { + var tags = EntityTags[(int)entity]; + tags.Add("result", ResultTag(outcome)); + + cycleDuration.Record(elapsed.TotalSeconds, tags); + + // A pass cut short by shutdown neither proves the sweep healthy nor faulty, so it leaves + // the gauge where it was. + if (outcome == CycleOutcome.Success) + { + Interlocked.Exchange(ref consecutiveFailures[(int)entity], 0); + } + else if (outcome == CycleOutcome.Failed) + { + 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 } }; + + static string ResultTag(CycleOutcome outcome) => outcome switch + { + CycleOutcome.Success => "success", + CycleOutcome.Cancelled => "cancelled", + CycleOutcome.Failed => "failed", + _ => throw new ArgumentOutOfRangeException(nameof(outcome)) + }; + + readonly long[] consecutiveFailures = new long[EntityTags.Length]; + + readonly Histogram cycleDuration; + + static readonly TagList[] EntityTags = + [ + EntityTag("audit_messages"), + EntityTag("saga_snapshots") + ]; + + const string MeterVersion = "0.1.0"; + const string InstrumentPrefix = "sc.retention"; +} diff --git a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs new file mode 100644 index 0000000000..e7fd82b690 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs @@ -0,0 +1,8 @@ +namespace ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + +enum CycleOutcome +{ + Success, + Cancelled, + Failed +} diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs new file mode 100644 index 0000000000..58e5c23bb8 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs @@ -0,0 +1,235 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + 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.Audit.Persistence.EFCore.Infrastructure.Metrics; + + /// + /// Instrument names are what dashboards and alerts are built on, so they are a published contract. + /// They match the error instance's retention instruments on purpose, so the same retention queries and alerts cover both instances, told apart by job. + /// The audit instance has no rows deleted counter, because PostgreSQL drops whole partitions and a dropped partition reports no row count. + /// + [TestFixture] + class AuditRetentionMetricsTests + { + [SetUp] + public void CreateMeterFactory() => provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + + [TearDown] + public void DisposeMeterFactory() => provider.Dispose(); + + [Test] + public void The_meter_publishes_the_error_instance_instruments_other_than_rows_deleted() + { + var published = new List(); + + var meterFactory = MeterFactory; + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, _) => + { + if (BelongsToThisTest(instrument, meterFactory)) + { + published.Add(instrument); + } + } + }; + + listener.Start(); + + _ = new AuditRetentionMetrics(MeterFactory); + + var cycleDuration = (Histogram)published.Single(instrument => instrument.Name == "sc.retention.cycle_duration_seconds"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(published.Select(instrument => instrument.Name).Order(), Is.EqualTo(new[] + { + "sc.retention.consecutive_failures_total", + "sc.retention.cycle_duration_seconds" + })); + Assert.That(published.Select(instrument => instrument.Meter.Name).Distinct(), Is.EqualTo(new[] { "Particular.ServiceControl.Audit" })); + Assert.That(cycleDuration.Advice?.HistogramBucketBoundaries, Is.EqualTo(new[] { 0.1, 0.5, 1, 5, 15, 60, 300, 900 })); + Assert.That(cycleDuration.Unit, Is.EqualTo("seconds")); + Assert.That(published.Single(instrument => instrument.Name == "sc.retention.consecutive_failures_total"), Is.InstanceOf>()); + } + } + + [Test] + public void A_completed_cycle_is_recorded_as_a_success() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + + using (var cycle = metrics.BeginCycle(AuditRetentionEntity.SagaSnapshots)) + { + cycle.Complete(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.Zero); + } + } + + [Test] + public void A_cycle_records_its_duration_in_seconds() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + + using (var cycle = metrics.BeginCycle(AuditRetentionEntity.AuditMessages)) + { + Thread.Sleep(50); + cycle.Complete(); + } + + Assert.That(recorded.Cycles(AuditRetentionEntity.AuditMessages)[0].Value, Is.InRange(0.04, 10)); + } + + [Test] + public void Every_kind_of_row_is_covered_by_a_tag() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + + Assert.That(() => + { + foreach (var entity in System.Enum.GetValues()) + { + metrics.BeginCycle(entity).Dispose(); + } + }, Throws.Nothing); + } + + [Test] + public void An_abandoned_cycle_is_recorded_as_a_failure() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(AuditRetentionEntity.SagaSnapshots).Dispose(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(1)); + } + } + + [Test] + public void Consecutive_failures_are_counted_per_entity() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(AuditRetentionEntity.AuditMessages).Dispose(); + metrics.BeginCycle(AuditRetentionEntity.AuditMessages).Dispose(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.EqualTo(2)); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.Zero); + } + } + + [Test] + public void A_success_clears_the_failures_of_that_entity_alone() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + + metrics.BeginCycle(AuditRetentionEntity.AuditMessages).Dispose(); + metrics.BeginCycle(AuditRetentionEntity.SagaSnapshots).Dispose(); + + using (var cycle = metrics.BeginCycle(AuditRetentionEntity.AuditMessages)) + { + cycle.Complete(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.Zero); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(1)); + } + } + + [Test] + public void A_cycle_interrupted_by_shutdown_is_recorded_as_cancelled() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + using (metrics.BeginCycle(AuditRetentionEntity.AuditMessages, shutdown.Token)) + { + shutdown.Cancel(); + } + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "cancelled" })); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.Zero); + } + } + + [Test] + public void A_cycle_that_finished_before_shutdown_is_still_a_success() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + using (var cycle = metrics.BeginCycle(AuditRetentionEntity.AuditMessages, shutdown.Token)) + { + cycle.Complete(); + shutdown.Cancel(); + } + + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "success" })); + } + + [Test] + public void Shutdown_neither_clears_nor_adds_to_the_failures() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + using var shutdown = new CancellationTokenSource(); + + metrics.BeginCycle(AuditRetentionEntity.AuditMessages).Dispose(); + + shutdown.Cancel(); + metrics.BeginCycle(AuditRetentionEntity.AuditMessages, shutdown.Token).Dispose(); + + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.EqualTo(1)); + } + + [Test] + public void Concurrent_failures_are_all_counted() + { + var metrics = new AuditRetentionMetrics(MeterFactory); + using var recorded = Listen(); + + const int failedCycles = 1000; + + Parallel.For(0, failedCycles, _ => metrics.BeginCycle(AuditRetentionEntity.AuditMessages).Dispose()); + + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.EqualTo(failedCycles)); + } + + RecordedAuditRetentionMetrics Listen() => new(MeterFactory); + + static bool BelongsToThisTest(Instrument instrument, IMeterFactory meterFactory) => + instrument.Meter.Name == AuditRetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, meterFactory); + + IMeterFactory MeterFactory => provider.GetRequiredService(); + + ServiceProvider provider; + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/BusyAuditPartitions.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/BusyAuditPartitions.cs new file mode 100644 index 0000000000..1f4e2986e0 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/BusyAuditPartitions.cs @@ -0,0 +1,31 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + using ServiceControl.Audit.Persistence.EFCore.DbContexts; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + + /// + /// Stands in for PostgreSQL when another session keeps a lock on an expired partition. + /// + sealed class BusyAuditPartitions : IAuditPartitionManager + { + static readonly DateTime ExpiredDay = new(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc); + + public Task EnsurePartitions(AuditDbContext dbContext, DateTime from, DateTime toExclusive, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default) => + Task.FromResult>([ExpiredDay]); + + public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime expired, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime start, int batchSize, CancellationToken cancellationToken = default) => + Task.FromResult(new DropResult(RowsDeleted: 0, Completed: true, Deferred: true)); + + public Task ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) => + Task.FromResult(null); + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionCustomCheckTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionCustomCheckTests.cs index 28e3f996f7..d30591f300 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionCustomCheckTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionCustomCheckTests.cs @@ -1,16 +1,15 @@ namespace ServiceControl.Audit.Persistence.Tests { using System; - using System.Collections.Generic; using System.Linq; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.DependencyInjection; using NServiceBus.CustomChecks; using NUnit.Framework; - using ServiceControl.Audit.Persistence.EFCore.DbContexts; using ServiceControl.Audit.Persistence.EFCore.Implementation; using ServiceControl.Audit.Persistence.EFCore.Infrastructure; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; class EFRetentionCustomCheckTests : EFPersistenceTestFixture { @@ -38,7 +37,7 @@ public async Task Check_fails_once_three_consecutive_sweeps_have_failed() [Test] public async Task Check_fails_once_three_consecutive_sweeps_have_left_expired_rows_behind() { - var retention = ActivatorUtilities.CreateInstance(ServiceProvider, new BusyPartitions()); + var retention = ActivatorUtilities.CreateInstance(ServiceProvider, new BusyAuditPartitions()); await retention.SweepNow(); await retention.SweepNow(); @@ -54,6 +53,27 @@ public async Task Check_fails_once_three_consecutive_sweeps_have_left_expired_ro } } + [TestCase(AuditRetentionEntity.AuditMessages)] + [TestCase(AuditRetentionEntity.SagaSnapshots)] + public async Task Check_fails_once_three_consecutive_sweeps_have_had_a_failing_pass(AuditRetentionEntity failing) + { + var partitions = new FailingAuditPartitions(failing, ServiceProvider.GetRequiredService()); + var retention = ActivatorUtilities.CreateInstance(ServiceProvider, partitions); + + await retention.SweepNow(); + await retention.SweepNow(); + var afterTwo = await Check.PerformCheck(); + await retention.SweepNow(); + var afterThree = await Check.PerformCheck(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(afterTwo.HasFailed, Is.False); + Assert.That(afterThree.HasFailed, Is.True); + Assert.That(afterThree.FailureReason, Does.Contain(FailingAuditPartitions.Reason)); + } + } + [Test] public async Task A_successful_sweep_clears_the_failure() { @@ -97,24 +117,5 @@ sealed class UnreachableRetentionLock : IRetentionLock public Task TryAcquire(CancellationToken cancellationToken = default) => Task.FromException(new InvalidOperationException(Reason)); } - - // Has one expired day that can never be dropped, like a partition another session keeps locked. - sealed class BusyPartitions : IAuditPartitionManager - { - static readonly DateTime ExpiredDay = new(2020, 1, 1, 0, 0, 0, DateTimeKind.Utc); - - public Task EnsurePartitions(AuditDbContext dbContext, DateTime from, DateTime toExclusive, CancellationToken cancellationToken = default) => Task.CompletedTask; - - public Task> ListExpired(AuditDbContext dbContext, DateTime keepFrom, CancellationToken cancellationToken = default) => - Task.FromResult>([ExpiredDay]); - - public Task DropExpired(AuditDbContext dbContext, DateTime start, int batchSize, CancellationToken cancellationToken = default) => - Task.FromResult(new DropResult(RowsDeleted: 0, Completed: true, Deferred: true)); - - public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime expired, CancellationToken cancellationToken = default) => Task.CompletedTask; - - public Task ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) => - Task.FromResult(null); - } } } diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionMetricsTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionMetricsTests.cs new file mode 100644 index 0000000000..5df547c0c8 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionMetricsTests.cs @@ -0,0 +1,271 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Diagnostics.Metrics; + using System.Linq; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.EntityFrameworkCore; + using Microsoft.Extensions.DependencyInjection; + using NServiceBus.CustomChecks; + using NUnit.Framework; + using ServiceControl.Audit.Persistence.EFCore.DbContexts; + using ServiceControl.Audit.Persistence.EFCore.Implementation; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + using ServiceControl.SagaAudit; + + class EFRetentionMetricsTests : EFPersistenceTestFixture + { + public override Task Setup() + { + SetSettings = settings => settings.AuditRetentionPeriod = TimeSpan.FromDays(1); + return base.Setup(); + } + + [Test] + public async Task A_sweep_reports_one_successful_pass_for_each_kind_of_row() + { + await Ingest(MakeMessage()); + await IngestSagaSnapshot(); + configuration.TimeProvider.Advance(TimeSpan.FromDays(3)); + using var recorded = Listen(); + + await configuration.CreateRetention().SweepNow(); + + var messagesLeft = await WithDbContext((dbContext, token) => dbContext.AuditMessages.CountAsync(token)); + var snapshotsLeft = await WithDbContext((dbContext, token) => dbContext.SagaSnapshots.CountAsync(token)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "success" })); + Assert.That(messagesLeft, Is.Zero, "the pass reported success without removing the expired messages"); + Assert.That(snapshotsLeft, Is.Zero, "the pass reported success without removing the expired snapshots"); + } + } + + [Test] + public async Task A_successful_sweep_clears_the_failures_a_failed_one_left() + { + using var recorded = Listen(); + await ActivatorUtilities.CreateInstance(ServiceProvider, new BusyAuditPartitions()).SweepNow(); + var afterFailure = recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages); + + await configuration.CreateRetention().SweepNow(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(afterFailure, Is.EqualTo(1)); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.Zero); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.Zero); + } + } + + [TestCase(AuditRetentionEntity.AuditMessages, AuditRetentionEntity.SagaSnapshots)] + [TestCase(AuditRetentionEntity.SagaSnapshots, AuditRetentionEntity.AuditMessages)] + public async Task A_failing_pass_does_not_stop_the_other_kind_of_row_from_being_removed(AuditRetentionEntity failing, AuditRetentionEntity other) + { + await Ingest(MakeMessage()); + await IngestSagaSnapshot(); + configuration.TimeProvider.Advance(TimeSpan.FromDays(3)); + using var recorded = Listen(); + var partitions = new FailingAuditPartitions(failing, ServiceProvider.GetRequiredService()); + + await ActivatorUtilities.CreateInstance(ServiceProvider, partitions).SweepNow(); + + var messagesLeft = await WithDbContext((dbContext, token) => dbContext.AuditMessages.CountAsync(token)); + var snapshotsLeft = await WithDbContext((dbContext, token) => dbContext.SagaSnapshots.CountAsync(token)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(failing), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.CycleResults(other), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.ConsecutiveFailures(failing), Is.EqualTo(1)); + Assert.That(recorded.ConsecutiveFailures(other), Is.Zero); + Assert.That(failing == AuditRetentionEntity.AuditMessages ? messagesLeft : snapshotsLeft, Is.EqualTo(1), "the other pass also removed this table's rows"); + Assert.That(other == AuditRetentionEntity.AuditMessages ? messagesLeft : snapshotsLeft, Is.Zero, "the failing pass stopped the other pass"); + } + } + + [Test] + public async Task A_pass_that_leaves_expired_rows_because_its_table_is_busy_is_reported_as_failed() + { + using var recorded = Listen(); + + await ActivatorUtilities.CreateInstance(ServiceProvider, new BusyAuditPartitions()).SweepNow(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.EqualTo(1)); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(1)); + } + } + + [Test] + public async Task A_sweep_that_fails_before_its_passes_reports_every_pass_as_failed() + { + using var recorded = Listen(); + var retention = ActivatorUtilities.CreateInstance(ServiceProvider, new UnreachableRetentionLock()); + + Assert.ThrowsAsync(() => retention.SweepNow()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.EqualTo(1)); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(1)); + } + } + + [Test] + public async Task A_sweep_whose_partition_provisioning_fails_reports_every_pass_as_failed() + { + using var recorded = Listen(); + var partitions = new UnprovisionablePartitions(ServiceProvider.GetRequiredService()); + var retention = ActivatorUtilities.CreateInstance(ServiceProvider, partitions); + + Assert.ThrowsAsync(() => retention.SweepNow()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "failed" })); + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "failed" })); + } + } + + [Test] + public async Task A_failure_after_the_passes_ran_does_not_record_them_again() + { + using var recorded = Listen(); + var retention = ActivatorUtilities.CreateInstance(ServiceProvider, new LockThatFailsToRelease()); + + Assert.ThrowsAsync(() => retention.SweepNow()); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "success" })); + Assert.That(recorded.CycleResults(AuditRetentionEntity.SagaSnapshots), Is.EqualTo(new[] { "success" })); + } + } + + [Test] + public async Task A_sweep_cut_short_by_shutdown_is_cancelled_rather_than_failed() + { + using var recorded = Listen(); + var check = ServiceProvider.GetServices().OfType().Single(); + var partitions = new ShutdownDuringFirstPass(ServiceProvider.GetRequiredService()); + + for (var sweep = 0; sweep < 3; sweep++) + { + using var shutdown = new CancellationTokenSource(); + partitions.Shutdown = shutdown; + var retention = ActivatorUtilities.CreateInstance(ServiceProvider, partitions); + + Assert.CatchAsync(() => retention.SweepNow(shutdown.Token)); + } + + var result = await check.PerformCheck(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(recorded.CycleResults(AuditRetentionEntity.AuditMessages), Is.EqualTo(new[] { "cancelled", "cancelled", "cancelled" })); + Assert.That(recorded.Cycles(AuditRetentionEntity.SagaSnapshots), Is.Empty, "the pass after shutdown still ran"); + Assert.That(recorded.ConsecutiveFailures(AuditRetentionEntity.AuditMessages), Is.Zero); + Assert.That(result.HasFailed, Is.False, "shutdown counted as a failed sweep"); + } + } + + [Test] + public async Task A_sweep_skipped_for_the_lock_reports_nothing() + { + await Ingest(MakeMessage()); + configuration.TimeProvider.Advance(TimeSpan.FromDays(3)); + using var recorded = Listen(); + + await using (await ServiceProvider.GetRequiredService().TryAcquire()) + { + await configuration.CreateRetention().SweepNow(); + } + + var whileLocked = recorded.RecordedAnything(); + await configuration.CreateRetention().SweepNow(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(whileLocked, Is.False); + Assert.That(recorded.RecordedAnything(), Is.True, "the recorder sees nothing at all, so the first assertion proves nothing"); + } + } + + async Task IngestSagaSnapshot() + { + await using var unitOfWork = await StartAuditUnitOfWork(1); + await unitOfWork.RecordSagaSnapshot(new SagaSnapshot { SagaId = Guid.NewGuid(), Status = SagaStateChangeStatus.New }); + await unitOfWork.Complete(); + await configuration.CompleteDBOperation(); + } + + RecordedAuditRetentionMetrics Listen() => new(ServiceProvider.GetRequiredService()); + + sealed class UnreachableRetentionLock : IRetentionLock + { + public Task TryAcquire(CancellationToken cancellationToken = default) => + Task.FromException(new InvalidOperationException("The database is unreachable")); + } + + sealed class LockThatFailsToRelease : IRetentionLock + { + public Task TryAcquire(CancellationToken cancellationToken = default) => Task.FromResult(new Handle()); + + sealed class Handle : IAsyncDisposable + { + public ValueTask DisposeAsync() => ValueTask.FromException(new InvalidOperationException("The lock connection was dropped")); + } + } + + sealed class UnprovisionablePartitions(IAuditPartitionManager inner) : IAuditPartitionManager + { + public Task EnsurePartitions(AuditDbContext dbContext, DateTime from, DateTime toExclusive, CancellationToken cancellationToken = default) => + Task.FromException(new InvalidOperationException("The partitions could not be created")); + + public Task> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default) => + inner.ListExpired(dbContext, entity, keepFrom, cancellationToken); + + public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime expired, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime start, int batchSize, CancellationToken cancellationToken = default) => + inner.DropExpired(dbContext, entity, start, batchSize, cancellationToken); + + public Task ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) => + inner.ProvisionedUntil(dbContext, cancellationToken); + } + + // Requests shutdown while the first pass lists its rows, the way a host stopping mid-sweep would. + sealed class ShutdownDuringFirstPass(IAuditPartitionManager inner) : IAuditPartitionManager + { + public CancellationTokenSource Shutdown { get; set; } + + public Task EnsurePartitions(AuditDbContext dbContext, DateTime from, DateTime toExclusive, CancellationToken cancellationToken = default) => + inner.EnsurePartitions(dbContext, from, toExclusive, cancellationToken); + + public Task> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default) + { + Shutdown.Cancel(); + return Task.FromCanceled>(cancellationToken); + } + + public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime expired, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime start, int batchSize, CancellationToken cancellationToken = default) => + inner.DropExpired(dbContext, entity, start, batchSize, cancellationToken); + + public Task ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) => + inner.ProvisionedUntil(dbContext, cancellationToken); + } + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs index eb2841d8ee..cf6b5c8ed8 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionTests.cs @@ -8,6 +8,7 @@ namespace ServiceControl.Audit.Persistence.Tests using NServiceBus; using NUnit.Framework; using ServiceControl.Audit.Infrastructure; + using ServiceControl.Audit.Persistence.EFCore.Implementation; using ServiceControl.Audit.Persistence.EFCore.Infrastructure; using ServiceControl.SagaAudit; @@ -80,7 +81,8 @@ public async Task Keeps_the_rows_when_deleting_their_bodies_fails() configuration.TimeProvider.Advance(TimeSpan.FromDays(3)); var storage = new RecordingBodyStorage { DeleteFailure = new IOException("storage is down") }; - Assert.ThrowsAsync(() => configuration.CreateRetention(storage).SweepNow()); + // A failed body deletion fails the pass, not the sweep: the rows stay for the next sweep to retry. + await configuration.CreateRetention(storage).SweepNow(); var remaining = await MessagesViewStore.GetMessages(true, new PagingInfo(), new SortInfo("time_sent", "desc")); var body = await MessagesViewStore.GetMessageBody(remaining.Results.Single().BodyUrl.Split('/')[2], TestContext.CurrentContext.CancellationToken); @@ -89,6 +91,7 @@ public async Task Keeps_the_rows_when_deleting_their_bodies_fails() { Assert.That(remaining.Results, Has.Count.EqualTo(1)); Assert.That(body.HasContent, Is.True, "the body is still in storage"); + Assert.That(ServiceProvider.GetRequiredService().Current.ConsecutiveFailedSweeps, Is.EqualTo(1), "the failed pass is recorded against the check"); } } diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/FailingAuditPartitions.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/FailingAuditPartitions.cs new file mode 100644 index 0000000000..3425c738f1 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/FailingAuditPartitions.cs @@ -0,0 +1,34 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Threading; + using System.Threading.Tasks; + using ServiceControl.Audit.Persistence.EFCore.DbContexts; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + + /// + /// Fails to list the expired rows of one kind of row, and hands everything else to the real partition manager. + /// + sealed class FailingAuditPartitions(AuditRetentionEntity failing, IAuditPartitionManager inner) : IAuditPartitionManager + { + public const string Reason = "The table is unreachable"; + + public Task EnsurePartitions(AuditDbContext dbContext, DateTime from, DateTime toExclusive, CancellationToken cancellationToken = default) => + inner.EnsurePartitions(dbContext, from, toExclusive, cancellationToken); + + public Task> ListExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime keepFrom, CancellationToken cancellationToken = default) => + entity == failing + ? Task.FromException>(new InvalidOperationException(Reason)) + : inner.ListExpired(dbContext, entity, keepFrom, cancellationToken); + + public Task DeleteBodies(IAuditBodyStoragePersistence bodyStorage, DateTime expired, CancellationToken cancellationToken = default) => Task.CompletedTask; + + public Task DropExpired(AuditDbContext dbContext, AuditRetentionEntity entity, DateTime start, int batchSize, CancellationToken cancellationToken = default) => + inner.DropExpired(dbContext, entity, start, batchSize, cancellationToken); + + public Task ProvisionedUntil(AuditDbContext dbContext, CancellationToken cancellationToken = default) => + inner.ProvisionedUntil(dbContext, cancellationToken); + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/OptionalBodyStorageRegistrationTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/OptionalBodyStorageRegistrationTests.cs index ecd37f16c8..73f0de7e50 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/EFCore/OptionalBodyStorageRegistrationTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/OptionalBodyStorageRegistrationTests.cs @@ -3,6 +3,7 @@ namespace ServiceControl.Audit.Persistence.Tests using System; using System.Linq; using Microsoft.Extensions.DependencyInjection; + using Microsoft.Extensions.Diagnostics.Metrics; using NServiceBus.CustomChecks; using NUnit.Framework; using ServiceControl.Audit.Persistence.EFCore.Abstractions; @@ -10,6 +11,7 @@ namespace ServiceControl.Audit.Persistence.Tests using ServiceControl.Audit.Persistence.EFCore.Implementation.BodyStorage; using ServiceControl.Audit.Persistence.EFCore.Implementation.UnitOfWork; using ServiceControl.Audit.Persistence.EFCore.Infrastructure; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; using ServiceControl.Audit.Persistence.UnitOfWork; [TestFixture] @@ -43,6 +45,8 @@ public void Consumers_resolve_without_a_body_store() services.AddSingleton(_ => null!); services.AddSingleton(_ => null!); services.AddSingleton(); + services.AddMetrics(); + services.AddSingleton(); services.AddLogging(); services.AddSingleton(); services.AddSingleton(); diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/RecordedAuditRetentionMetrics.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/RecordedAuditRetentionMetrics.cs new file mode 100644 index 0000000000..a1613531c1 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/RecordedAuditRetentionMetrics.cs @@ -0,0 +1,98 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Diagnostics.Metrics; + using System.Linq; + using ServiceControl.Audit.Persistence.EFCore.Infrastructure.Metrics; + + /// + /// Collects everything the audit 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 RecordedAuditRetentionMetrics : IDisposable + { + public RecordedAuditRetentionMetrics(IMeterFactory meterFactory) + { + listener = new MeterListener + { + InstrumentPublished = (instrument, activeListener) => + { + if (instrument.Meter.Name == AuditRetentionMetrics.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, AuditRetentionEntity entity) + { + lock (measurements) + { + return + [ + .. measurements.Where(measurement => + measurement.InstrumentName == instrumentName && + Equals(measurement.Tags["retention.entity"], EntityTag(entity))) + ]; + } + } + + public IReadOnlyList Cycles(AuditRetentionEntity entity) => Of(AuditRetentionMetrics.CycleDurationInstrumentName, entity); + + public IReadOnlyList CycleResults(AuditRetentionEntity entity) => [.. Cycles(entity).Select(cycle => cycle.Result)]; + + public double ConsecutiveFailures(AuditRetentionEntity entity) + { + listener.RecordObservableInstruments(); + + return Of(AuditRetentionMetrics.ConsecutiveFailuresInstrumentName, entity)[^1].Value; + } + + public bool RecordedAnything() + { + lock (measurements) + { + return measurements.Any(measurement => measurement.InstrumentName != AuditRetentionMetrics.ConsecutiveFailuresInstrumentName); + } + } + + 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(AuditRetentionEntity entity) => entity switch + { + AuditRetentionEntity.AuditMessages => "audit_messages", + AuditRetentionEntity.SagaSnapshots => "saga_snapshots", + _ => throw new ArgumentOutOfRangeException(nameof(entity)) + }; + + readonly MeterListener listener; + readonly List measurements = []; + + public sealed record Recorded(string InstrumentName, double Value, IReadOnlyDictionary Tags) + { + public string Result => (string)Tags["result"]; + } + } +} diff --git a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs index 2afe91ec6a..53210793f6 100644 --- a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs +++ b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs @@ -3,6 +3,7 @@ namespace ServiceControl.Transport.Tests; using System; using System.Collections.Generic; using System.Collections.ObjectModel; +using System.Diagnostics; using System.Linq; using System.Text.Json.Nodes; using System.Text.RegularExpressions; @@ -212,18 +213,23 @@ public async Task RunScenario() IBrokerQueue queue = queueNames.Find(name => name.QueueName == transportSettings.EndpointName); Assert.That(queue, Is.Not.Null); - long total = 0L; - - await Task.Delay(TimeSpan.FromMinutes(4), token); - DateTime startDate = provider.GetUtcNow().DateTime; provider.Advance(TimeSpan.FromDays(1)); - await foreach (QueueThroughput queueThroughput in query.GetThroughputPerDay(queue, DateOnly.FromDateTime(startDate), token)) + + long total = 0L; + var waited = Stopwatch.StartNew(); + while (total < numMessagesToIngest && waited.Elapsed < TimeSpan.FromMinutes(9)) { - total += queueThroughput.TotalThroughput; + await Task.Delay(TimeSpan.FromSeconds(15), token); + + total = 0L; + await foreach (QueueThroughput queueThroughput in query.GetThroughputPerDay(queue, DateOnly.FromDateTime(startDate), token)) + { + total += queueThroughput.TotalThroughput; + } } - Assert.That(total, Is.EqualTo(numMessagesToIngest)); + Assert.That(total, Is.EqualTo(numMessagesToIngest), $"Azure Monitor throughput after waiting {waited.Elapsed:mm\\:ss}"); } static Dictionary GetSettings() diff --git a/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs b/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs index 86560cd816..eb1c23fd44 100644 --- a/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs +++ b/src/ServiceControl.UnitTests/Operations/Metrics/IngestionMetricsTests.cs @@ -28,11 +28,13 @@ public void The_meter_publishes_the_instruments_it_is_named_for() { var published = new List(); + var meterFactory = MeterFactory; + using var listener = new MeterListener { InstrumentPublished = (instrument, _) => { - if (BelongsToThisTest(instrument)) + if (BelongsToThisTest(instrument, meterFactory)) { published.Add(instrument.Name); } @@ -72,11 +74,13 @@ long ReadConsecutiveBatchFailures() { long value = -1; + var meterFactory = MeterFactory; + using var listener = new MeterListener { InstrumentPublished = (instrument, activeListener) => { - if (BelongsToThisTest(instrument) && instrument.Name == "sc.error.ingestion.consecutive_batch_failures_total") + if (BelongsToThisTest(instrument, meterFactory) && instrument.Name == "sc.error.ingestion.consecutive_batch_failures_total") { activeListener.EnableMeasurementEvents(instrument); } @@ -90,10 +94,8 @@ long ReadConsecutiveBatchFailures() 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); + static bool BelongsToThisTest(Instrument instrument, IMeterFactory meterFactory) => + instrument.Meter.Name == IngestionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, meterFactory); IMeterFactory MeterFactory => provider.GetRequiredService(); diff --git a/src/ServiceControl.UnitTests/Recoverability/Metrics/RetryMetricsTests.cs b/src/ServiceControl.UnitTests/Recoverability/Metrics/RetryMetricsTests.cs index 9dada2d37a..d214fccfa7 100644 --- a/src/ServiceControl.UnitTests/Recoverability/Metrics/RetryMetricsTests.cs +++ b/src/ServiceControl.UnitTests/Recoverability/Metrics/RetryMetricsTests.cs @@ -34,11 +34,13 @@ public void The_meter_publishes_the_instruments_it_is_named_for() { var published = new List(); + var meterFactory = MeterFactory; + using var listener = new MeterListener { InstrumentPublished = (instrument, _) => { - if (BelongsToThisTest(instrument)) + if (BelongsToThisTest(instrument, meterFactory)) { published.Add(instrument.Name); } @@ -215,10 +217,8 @@ public void The_bulk_request_gauge_reads_the_queue_depth() Assert.That(recorded.Observe(RetryMetrics.PendingBulkRequestsInstrumentName).Single().Value, Is.Zero); } - // 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 == RetryMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory); + static bool BelongsToThisTest(Instrument instrument, IMeterFactory meterFactory) => + instrument.Meter.Name == RetryMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, meterFactory); IMeterFactory MeterFactory => provider.GetRequiredService(); diff --git a/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json b/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json index e7bf4ac347..bdeda1db03 100644 --- a/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json +++ b/tools/testing-tool/TestingTool.AppHost/obs/grafana/dashboards/testing-tool.json @@ -1386,7 +1386,7 @@ { "id": 35, "title": "Recoverability Throughput / sec (messages moved)", - "description": "Combined view of items moving through the three recoverability pipelines: sc.retry.messages_total (messages staged/forwarded/skipped by retry type), sc.archive.messages_total (messages archived/unarchived), and sc.retention.rows_deleted_total (rows deleted by entity). Stacked, the total is the rate of recoverability work ServiceControl is doing. Compare against the testing tool's own errors_replayed_total and errors_archived_total to see whether ServiceControl is keeping up with the requested work.", + "description": "Combined view of items moving through the three recoverability pipelines: sc.retry.messages_total (messages staged/forwarded/skipped by retry type), sc.archive.messages_total (messages archived/unarchived), and sc.retention.rows_deleted_total (rows deleted by entity, error instance only). Stacked, the total is the rate of recoverability work ServiceControl is doing. Compare against the testing tool's own errors_replayed_total and errors_archived_total to see whether ServiceControl is keeping up with the requested work.", "type": "timeseries", "gridPos": { "h": 8, @@ -1539,7 +1539,7 @@ { "id": 38, "title": "Archive & Retention Durations (s) \u2014 mean", - "description": "Mean duration for the remaining recoverability pipelines: sc.archive.operation_duration_seconds (full archive/unarchive operation), sc.archive.batch_duration_seconds (individual archive batch), and sc.retention.cycle_duration_seconds (retention sweep pass). True means (histogram sum / count). Archive operation duration spans the whole group; batch duration is the per-batch write. Retention cycle is the full sweep pass, split by entity in the legend.", + "description": "Mean duration for the remaining recoverability pipelines: sc.archive.operation_duration_seconds (full archive/unarchive operation), sc.archive.batch_duration_seconds (individual archive batch), and sc.retention.cycle_duration_seconds (retention sweep pass). True means (histogram sum / count). Archive operation duration spans the whole group; batch duration is the per-batch write. Retention cycle is the full sweep pass, averaged since the instance started because the sweep runs only hourly, and split by entity in the legend: failed_messages, event_log and group_comments from the error instance, audit_messages and saga_snapshots from the audit instance.", "type": "timeseries", "gridPos": { "h": 8, @@ -1581,7 +1581,7 @@ "legendFormat": "archive batch" }, { - "expr": "sum by (retention_entity) (rate(sc_retention_cycle_duration_seconds_sum{exported_job=~\"$sc_instance\"}[$__rate_interval])) / sum by (retention_entity) (rate(sc_retention_cycle_duration_seconds_count{exported_job=~\"$sc_instance\"}[$__rate_interval]))", + "expr": "sum by (retention_entity) (sc_retention_cycle_duration_seconds_sum{exported_job=~\"$sc_instance\"}) / sum by (retention_entity) (sc_retention_cycle_duration_seconds_count{exported_job=~\"$sc_instance\"})", "legendFormat": "retention: {{retention_entity}}" } ] @@ -1589,7 +1589,7 @@ { "id": 39, "title": "Consecutive Retention Failures", - "description": "sc.retention.consecutive_failures_total \u2014 the retention sweep's own circuit-breaker signal, split by entity (failed_messages, event_log, group_comments). Resets to 0 on the first successful sweep. Anything above 0 means the sweep is currently failing. Mirrors the ingestion 'Consecutive Batch Failures' panel (id 23).", + "description": "sc.retention.consecutive_failures_total: the retention sweep's own circuit-breaker signal, split by entity (failed_messages, event_log and group_comments on the error instance, audit_messages and saga_snapshots on the audit instance). Resets to 0 on the first successful sweep. Anything above 0 means the sweep is currently failing. Mirrors the ingestion 'Consecutive Batch Failures' panel (id 23).", "type": "stat", "gridPos": { "h": 6, @@ -1697,6 +1697,78 @@ "legendFormat": "pending" } ] + }, + { + "id": 43, + "title": "Retention Passes Since Start (by entity and result)", + "description": "Retention sweep passes counted from sc.retention.cycle_duration_seconds since the instance started, split by entity and result (success, failed, cancelled). The sweep runs one minute after start and then hourly, so a healthy instance gains one success per entity each hour. On the audit instance, a pass that leaves expired rows because its table is too busy records failed. An audit instance that did not hold the retention lock records nothing.", + "type": "stat", + "gridPos": { + "h": 6, + "w": 24, + "x": 0, + "y": 114 + }, + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "fieldConfig": { + "defaults": { + "unit": "short", + "color": { + "mode": "fixed", + "fixedColor": "green" + } + }, + "overrides": [ + { + "matcher": { + "id": "byRegexp", + "options": ".* failed$" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "red" + } + } + ] + }, + { + "matcher": { + "id": "byRegexp", + "options": ".* cancelled$" + }, + "properties": [ + { + "id": "color", + "value": { + "mode": "fixed", + "fixedColor": "orange" + } + } + ] + } + ] + }, + "options": { + "reduceOptions": { + "calcs": [ + "lastNotNull" + ] + }, + "graphMode": "none", + "colorMode": "value" + }, + "targets": [ + { + "expr": "sum by (retention_entity, result) (sc_retention_cycle_duration_seconds_count{exported_job=~\"$sc_instance\"})", + "legendFormat": "{{retention_entity}} {{result}}" + } + ] } ] }