From 6e1e8dfa6acd5c749c267293b99c49b6739c0988 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Wed, 28 Oct 2026 15:13:08 +0800 Subject: [PATCH 1/8] feat: Implement audit retention metrics and improve retention sweep logic. --- docs/telemetry.md | 15 +- .../PostgreSqlAuditPartitionManager.cs | 28 +- .../SqlServerAuditPartitionManager.cs | 47 +-- .../Abstractions/EFPersistenceBase.cs | 2 + .../Implementation/AuditRetention.cs | 80 +++++- .../Infrastructure/IAuditPartitionManager.cs | 5 +- .../Metrics/AuditRetentionCycleMetrics.cs | 40 +++ .../Metrics/AuditRetentionEntity.cs | 7 + .../Metrics/AuditRetentionMetrics.cs | 81 ++++++ .../Infrastructure/Metrics/CycleOutcome.cs | 8 + .../EFCore/AuditRetentionMetricsTests.cs | 235 +++++++++++++++ .../EFCore/BusyAuditPartitions.cs | 29 ++ .../EFCore/EFRetentionCustomCheckTests.cs | 43 +-- .../EFCore/EFRetentionMetricsTests.cs | 267 ++++++++++++++++++ .../EFCore/FailingAuditPartitions.cs | 32 +++ .../EFCore/RecordedAuditRetentionMetrics.cs | 98 +++++++ 16 files changed, 937 insertions(+), 80 deletions(-) create mode 100644 src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionCycleMetrics.cs create mode 100644 src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionEntity.cs create mode 100644 src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/AuditRetentionMetrics.cs create mode 100644 src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/Metrics/CycleOutcome.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests/EFCore/BusyAuditPartitions.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionMetricsTests.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests/EFCore/FailingAuditPartitions.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests/EFCore/RecordedAuditRetentionMetrics.cs 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 1c310dd52f..52f508fc71 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,26 +36,16 @@ 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()]; } - 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) @@ -118,6 +109,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 65e436b7f1..602d2f1cfb 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,34 +11,36 @@ 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 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 90cb94678b..0b4f23fdf9 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore/Abstractions/EFPersistenceBase.cs +++ b/src/ServiceControl.Audit.Persistence.EFCore/Abstractions/EFPersistenceBase.cs @@ -5,6 +5,7 @@ namespace ServiceControl.Audit.Persistence.EFCore.Abstractions; using NServiceBus; using ServiceControl.Audit.Persistence.EFCore.Implementation; 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 @@ -23,6 +24,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 588631ee32..b7d5402369 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) : BackgroundService { @@ -67,9 +69,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) @@ -78,13 +81,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 { @@ -94,34 +107,71 @@ 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)) { 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; } @@ -131,9 +181,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 2b6e7de104..4c32cca4cc 100644 --- a/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/IAuditPartitionManager.cs +++ b/src/ServiceControl.Audit.Persistence.EFCore/Infrastructure/IAuditPartitionManager.cs @@ -1,14 +1,15 @@ 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); - 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..21b181b35d --- /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(); + + using var listener = new MeterListener + { + InstrumentPublished = (instrument, _) => + { + if (BelongsToThisTest(instrument)) + { + 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); + + // 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 == 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..0dc95feb43 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/BusyAuditPartitions.cs @@ -0,0 +1,29 @@ +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 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 fa342f4a46..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,22 +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 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..abce60e062 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/EFRetentionMetricsTests.cs @@ -0,0 +1,267 @@ +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 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 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/FailingAuditPartitions.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/FailingAuditPartitions.cs new file mode 100644 index 0000000000..8c12ade6a4 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/FailingAuditPartitions.cs @@ -0,0 +1,32 @@ +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 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/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"]; + } + } +} From 9cb260d7754e433c5b9af083e280e27950e79869 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Wed, 28 Oct 2026 15:39:51 +0800 Subject: [PATCH 2/8] Fix flakey metrics test --- .../EFCore/AuditRetentionMetricsTests.cs | 10 +++++----- .../Operations/Metrics/IngestionMetricsTests.cs | 14 ++++++++------ .../Recoverability/Metrics/RetryMetricsTests.cs | 10 +++++----- 3 files changed, 18 insertions(+), 16 deletions(-) diff --git a/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs b/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs index 21b181b35d..58e5c23bb8 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/EFCore/AuditRetentionMetricsTests.cs @@ -28,11 +28,13 @@ public void The_meter_publishes_the_error_instance_instruments_other_than_rows_d { var published = new List(); + var meterFactory = MeterFactory; + using var listener = new MeterListener { InstrumentPublished = (instrument, _) => { - if (BelongsToThisTest(instrument)) + if (BelongsToThisTest(instrument, meterFactory)) { published.Add(instrument); } @@ -223,10 +225,8 @@ public void Concurrent_failures_are_all_counted() RecordedAuditRetentionMetrics Listen() => new(MeterFactory); - // Every fixture in the run shares the meter name, so the factory is what tells the instruments - // created here apart from the ones another test left behind. - bool BelongsToThisTest(Instrument instrument) => - instrument.Meter.Name == AuditRetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory); + static bool BelongsToThisTest(Instrument instrument, IMeterFactory meterFactory) => + instrument.Meter.Name == AuditRetentionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, meterFactory); IMeterFactory MeterFactory => provider.GetRequiredService(); 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(); From de94eb8ada15810e99131a7b0eb602624e9d6972 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Thu, 29 Oct 2026 13:58:37 +0800 Subject: [PATCH 3/8] feat: Update Grafana dashboard descriptions and add retention passes panel --- .../obs/grafana/dashboards/testing-tool.json | 80 ++++++++++++++++++- 1 file changed, 76 insertions(+), 4 deletions(-) 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}}" + } + ] } ] } From b5ee9d9795ee84ebefb0b057417031db13a38a1d Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Thu, 29 Oct 2026 15:09:26 +0800 Subject: [PATCH 4/8] feat: Add Azure Monitor namespace wait logic to AzureQueryTests --- .../AzureQueryTests.cs | 40 +++++++++++++++++++ 1 file changed, 40 insertions(+) diff --git a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs index 2afe91ec6a..9f80e954a1 100644 --- a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs +++ b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs @@ -3,11 +3,13 @@ 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; using System.Threading; using System.Threading.Tasks; +using Azure; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Time.Testing; using NUnit.Framework; @@ -23,6 +25,44 @@ class AzureQueryTests : TransportTestFixture TransportSettings transportSettings; AzureQuery query; + static readonly TimeSpan MonitorMappingTimeout = TimeSpan.FromMinutes(15); + static readonly TimeSpan MonitorMappingPollInterval = TimeSpan.FromSeconds(15); + + // Azure Monitor answers 404 for a namespace it has not picked up yet, and CI creates a new namespace for every run. + [OneTimeSetUp] + public async Task WaitUntilAzureMonitorKnowsTheNamespace() + { + var testsConfiguration = new TransportTestsConfiguration(); + await testsConfiguration.Configure(); + + var metricsQuery = new AzureQuery(NullLogger.Instance, TimeProvider.System, + new TransportSettings { ConnectionString = testsConfiguration.ConnectionString }); + metricsQuery.Initialize(new ReadOnlyDictionary(GetSettings())); + await metricsQuery.GetQueueNames().ToListAsync(); + + var yesterday = DateOnly.FromDateTime(TimeProvider.System.GetUtcNow().UtcDateTime).AddDays(-1); + var waited = Stopwatch.StartNew(); + while (true) + { + try + { + await metricsQuery.GetThroughputPerDay(new DefaultBrokerQueue("any"), yesterday).ToListAsync(); + break; + } + catch (RequestFailedException ex) when (ex.Status == 404 && ex.Message.Contains("Failed to find resource mapping")) + { + if (waited.Elapsed > MonitorMappingTimeout) + { + throw new Exception($"Azure Monitor still could not find the Service Bus namespace after {waited.Elapsed:mm\\:ss}.", ex); + } + + await Task.Delay(MonitorMappingPollInterval); + } + } + + await TestContext.Progress.WriteLineAsync($"Azure Monitor found the Service Bus namespace after {waited.Elapsed:mm\\:ss}."); + } + [SetUp] public void Initialise() { From 8b7fec640c17e9cc6f8887c7f57fdff9e2a4b711 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Thu, 29 Oct 2026 15:25:24 +0800 Subject: [PATCH 5/8] add Azure Monitor resource mapping probe to AzureQueryTests to debug flakey test --- .../AzureQueryTests.cs | 118 ++++++++++++------ 1 file changed, 79 insertions(+), 39 deletions(-) diff --git a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs index 9f80e954a1..9173f005a1 100644 --- a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs +++ b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs @@ -3,13 +3,19 @@ namespace ServiceControl.Transport.Tests; using System; using System.Collections.Generic; using System.Collections.ObjectModel; -using System.Diagnostics; using System.Linq; +using System.Text; using System.Text.Json.Nodes; using System.Text.RegularExpressions; using System.Threading; using System.Threading.Tasks; using Azure; +using Azure.Identity; +using Azure.ResourceManager; +using Azure.ResourceManager.Monitor; +using Azure.ResourceManager.Monitor.Models; +using Azure.ResourceManager.Resources; +using Azure.ResourceManager.ServiceBus; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Time.Testing; using NUnit.Framework; @@ -25,44 +31,6 @@ class AzureQueryTests : TransportTestFixture TransportSettings transportSettings; AzureQuery query; - static readonly TimeSpan MonitorMappingTimeout = TimeSpan.FromMinutes(15); - static readonly TimeSpan MonitorMappingPollInterval = TimeSpan.FromSeconds(15); - - // Azure Monitor answers 404 for a namespace it has not picked up yet, and CI creates a new namespace for every run. - [OneTimeSetUp] - public async Task WaitUntilAzureMonitorKnowsTheNamespace() - { - var testsConfiguration = new TransportTestsConfiguration(); - await testsConfiguration.Configure(); - - var metricsQuery = new AzureQuery(NullLogger.Instance, TimeProvider.System, - new TransportSettings { ConnectionString = testsConfiguration.ConnectionString }); - metricsQuery.Initialize(new ReadOnlyDictionary(GetSettings())); - await metricsQuery.GetQueueNames().ToListAsync(); - - var yesterday = DateOnly.FromDateTime(TimeProvider.System.GetUtcNow().UtcDateTime).AddDays(-1); - var waited = Stopwatch.StartNew(); - while (true) - { - try - { - await metricsQuery.GetThroughputPerDay(new DefaultBrokerQueue("any"), yesterday).ToListAsync(); - break; - } - catch (RequestFailedException ex) when (ex.Status == 404 && ex.Message.Contains("Failed to find resource mapping")) - { - if (waited.Elapsed > MonitorMappingTimeout) - { - throw new Exception($"Azure Monitor still could not find the Service Bus namespace after {waited.Elapsed:mm\\:ss}.", ex); - } - - await Task.Delay(MonitorMappingPollInterval); - } - } - - await TestContext.Progress.WriteLineAsync($"Azure Monitor found the Service Bus namespace after {waited.Elapsed:mm\\:ss}."); - } - [SetUp] public void Initialise() { @@ -77,6 +45,78 @@ public void Initialise() query = new AzureQuery(NullLogger.Instance, provider, transportSettings); } + // Temporary probe for the Azure Monitor 404 seen in CI. It always fails, because only a failure message reaches the CI log. + [Test, Order(0)] + public async Task ProbeAzureMonitorResourceMapping() + { + Dictionary settings = GetSettings(); + var credential = new ClientSecretCredential( + settings[AzureQuery.AzureServiceBusSettings.TenantId], + settings[AzureQuery.AzureServiceBusSettings.ClientId], + settings[AzureQuery.AzureServiceBusSettings.ClientSecret]); + var armClient = new ArmClient(credential, settings[AzureQuery.AzureServiceBusSettings.SubscriptionId]); + string namespaceName = query.ExtractServiceBusName(); + SubscriptionResource subscription = await armClient.GetDefaultSubscriptionAsync(); + ServiceBusNamespaceResource serviceBusNamespace = await subscription.GetServiceBusNamespacesAsync() + .SingleAsync(ns => string.Equals(ns.Data.Name, namespaceName, StringComparison.OrdinalIgnoreCase)); + string queueName = transportSettings.EndpointName; + + var log = new StringBuilder(); + log.AppendLine($"Namespace {serviceBusNamespace.Data.Name} in {serviceBusNamespace.Data.Location}, created {serviceBusNamespace.Data.CreatedOn:HH:mm:ss}"); + + (string Metric, string Filter)[] variants = + [ + ("CompleteMessage", "EntityName eq 'any'"), + ("CompleteMessage", $"EntityName eq '{queueName}'"), + ("CompleteMessage", null), + ("IncomingMessages", null) + ]; + + async Task Probe(string phase) + { + DateTime today = DateTime.UtcNow.Date; + foreach ((string metric, string filter) in variants) + { + var options = new ArmResourceGetMonitorMetricsOptions + { + Metricnames = metric, + Timespan = $"{today.AddDays(-1):o}/{today.AddDays(1):o}", + Filter = filter, + Interval = TimeSpan.FromDays(1), + Metricnamespace = "Microsoft.ServiceBus/Namespaces" + }; + + string outcome; + try + { + List metrics = await armClient.GetMonitorMetricsAsync(serviceBusNamespace.Id, options).ToListAsync(); + outcome = $"OK total={metrics.SelectMany(m => m.Timeseries).SelectMany(t => t.Data).Sum(d => d.Total ?? 0)}"; + } + catch (RequestFailedException ex) + { + outcome = $"{ex.Status} {ex.ErrorCode}: {ex.Message.Split('\n')[0]}"; + } + + log.AppendLine($"{DateTime.UtcNow:HH:mm:ss} {phase,-13} {metric,-16} {filter ?? "(no filter)",-52} {outcome}"); + } + } + + await Probe("empty"); + await CreateTestQueue(queueName); + await Probe("queue created"); + await SendAndReceiveMessages(queueName, 15); + await Probe("messages sent"); + + DateTime stopAt = DateTime.UtcNow.AddMinutes(7); + while (DateTime.UtcNow < stopAt) + { + await Task.Delay(TimeSpan.FromSeconds(30)); + await Probe("polling"); + } + + Assert.Fail($"Azure Monitor probe results:{Environment.NewLine}{log}"); + } + [Test] public async Task TestConnectionWithEmptySettings() { From ec7761b29d99126feea9ef5314cc3222276e5de4 Mon Sep 17 00:00:00 2001 From: Warwick Schroeder Date: Thu, 29 Oct 2026 15:51:18 +0800 Subject: [PATCH 6/8] refactor: Remove Azure Monitor probe and improve throughput calculation logic in AzureQueryTests --- .../AzureQueryTests.cs | 100 +++--------------- 1 file changed, 13 insertions(+), 87 deletions(-) diff --git a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs index 9173f005a1..53210793f6 100644 --- a/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs +++ b/src/ServiceControl.Transports.ASBS.Tests/AzureQueryTests.cs @@ -3,19 +3,12 @@ namespace ServiceControl.Transport.Tests; using System; using System.Collections.Generic; using System.Collections.ObjectModel; +using System.Diagnostics; using System.Linq; -using System.Text; using System.Text.Json.Nodes; using System.Text.RegularExpressions; using System.Threading; using System.Threading.Tasks; -using Azure; -using Azure.Identity; -using Azure.ResourceManager; -using Azure.ResourceManager.Monitor; -using Azure.ResourceManager.Monitor.Models; -using Azure.ResourceManager.Resources; -using Azure.ResourceManager.ServiceBus; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Time.Testing; using NUnit.Framework; @@ -45,78 +38,6 @@ public void Initialise() query = new AzureQuery(NullLogger.Instance, provider, transportSettings); } - // Temporary probe for the Azure Monitor 404 seen in CI. It always fails, because only a failure message reaches the CI log. - [Test, Order(0)] - public async Task ProbeAzureMonitorResourceMapping() - { - Dictionary settings = GetSettings(); - var credential = new ClientSecretCredential( - settings[AzureQuery.AzureServiceBusSettings.TenantId], - settings[AzureQuery.AzureServiceBusSettings.ClientId], - settings[AzureQuery.AzureServiceBusSettings.ClientSecret]); - var armClient = new ArmClient(credential, settings[AzureQuery.AzureServiceBusSettings.SubscriptionId]); - string namespaceName = query.ExtractServiceBusName(); - SubscriptionResource subscription = await armClient.GetDefaultSubscriptionAsync(); - ServiceBusNamespaceResource serviceBusNamespace = await subscription.GetServiceBusNamespacesAsync() - .SingleAsync(ns => string.Equals(ns.Data.Name, namespaceName, StringComparison.OrdinalIgnoreCase)); - string queueName = transportSettings.EndpointName; - - var log = new StringBuilder(); - log.AppendLine($"Namespace {serviceBusNamespace.Data.Name} in {serviceBusNamespace.Data.Location}, created {serviceBusNamespace.Data.CreatedOn:HH:mm:ss}"); - - (string Metric, string Filter)[] variants = - [ - ("CompleteMessage", "EntityName eq 'any'"), - ("CompleteMessage", $"EntityName eq '{queueName}'"), - ("CompleteMessage", null), - ("IncomingMessages", null) - ]; - - async Task Probe(string phase) - { - DateTime today = DateTime.UtcNow.Date; - foreach ((string metric, string filter) in variants) - { - var options = new ArmResourceGetMonitorMetricsOptions - { - Metricnames = metric, - Timespan = $"{today.AddDays(-1):o}/{today.AddDays(1):o}", - Filter = filter, - Interval = TimeSpan.FromDays(1), - Metricnamespace = "Microsoft.ServiceBus/Namespaces" - }; - - string outcome; - try - { - List metrics = await armClient.GetMonitorMetricsAsync(serviceBusNamespace.Id, options).ToListAsync(); - outcome = $"OK total={metrics.SelectMany(m => m.Timeseries).SelectMany(t => t.Data).Sum(d => d.Total ?? 0)}"; - } - catch (RequestFailedException ex) - { - outcome = $"{ex.Status} {ex.ErrorCode}: {ex.Message.Split('\n')[0]}"; - } - - log.AppendLine($"{DateTime.UtcNow:HH:mm:ss} {phase,-13} {metric,-16} {filter ?? "(no filter)",-52} {outcome}"); - } - } - - await Probe("empty"); - await CreateTestQueue(queueName); - await Probe("queue created"); - await SendAndReceiveMessages(queueName, 15); - await Probe("messages sent"); - - DateTime stopAt = DateTime.UtcNow.AddMinutes(7); - while (DateTime.UtcNow < stopAt) - { - await Task.Delay(TimeSpan.FromSeconds(30)); - await Probe("polling"); - } - - Assert.Fail($"Azure Monitor probe results:{Environment.NewLine}{log}"); - } - [Test] public async Task TestConnectionWithEmptySettings() { @@ -292,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() From 026c21598f6749b030958cf20d9ef72d4ef7907d Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Fri, 30 Oct 2026 15:38:29 +0800 Subject: [PATCH 7/8] Fix failing test --- .../EFCore/OptionalBodyStorageRegistrationTests.cs | 4 ++++ 1 file changed, 4 insertions(+) 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(); From b434d14a8bc52fad9e320b885029f789481035b7 Mon Sep 17 00:00:00 2001 From: Rhys Bevilaqua Date: Fri, 30 Oct 2026 15:53:39 +0800 Subject: [PATCH 8/8] Change test to support updated exception handling --- .../EFCore/EFRetentionTests.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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"); } }