From f33227e2085cd6bcbfd8c75cd2b40780296dae95 Mon Sep 17 00:00:00 2001 From: Ramon Smits Date: Fri, 30 Oct 2026 13:33:58 +0100 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Add=20--apply-retention=20to=20the?= =?UTF-8?q?=20audit=20instance?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A change of ServiceControl.Audit/AuditRetentionPeriod only reaches new messages: ingestion sets the expiry of each document when it is stored. The new command rewrites the expiry of the stored messages and saga snapshots to their processed time plus the configured retention period. - IAuditRetentionApplier in the persistence abstraction, registered only by a persister that stores an expiry per document - RavenDB: patch by query over ProcessedMessages and SagaSnapshots, progress reported per batch - Without a registered implementation the command reports that the persister applies the retention period when it removes expired messages - Help text, host arguments test, RavenDB persistence test --- .../RavenAuditRetentionApplier.cs | 59 +++++++++++++++ .../RavenPersistence.cs | 1 + .../ApplyRetentionTests.cs | 74 +++++++++++++++++++ .../IAuditRetentionApplier.cs | 20 +++++ .../Infrastructure/HostArgumentsTests.cs | 22 ++++++ .../Hosting/Commands/ApplyRetentionCommand.cs | 67 +++++++++++++++++ .../Infrastructure/Hosting/Help.txt | 13 ++++ .../Infrastructure/Hosting/HostArguments.cs | 16 ++++ 8 files changed, 272 insertions(+) create mode 100644 src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditRetentionApplier.cs create mode 100644 src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApplyRetentionTests.cs create mode 100644 src/ServiceControl.Audit.Persistence/IAuditRetentionApplier.cs create mode 100644 src/ServiceControl.Audit.UnitTests/Infrastructure/HostArgumentsTests.cs create mode 100644 src/ServiceControl.Audit/Infrastructure/Hosting/Commands/ApplyRetentionCommand.cs diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditRetentionApplier.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditRetentionApplier.cs new file mode 100644 index 0000000000..e7baf63512 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditRetentionApplier.cs @@ -0,0 +1,59 @@ +namespace ServiceControl.Audit.Persistence.RavenDB +{ + using System; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.Logging; + using Raven.Client; + using Raven.Client.Documents.Operations; + using Raven.Client.Documents.Queries; + + class RavenAuditRetentionApplier(IRavenDocumentStoreProvider documentStoreProvider, ILogger logger) : IAuditRetentionApplier + { + // Ingestion sets the expiry from the ingestion time. That time is not stored; ProcessedAt is the stored time closest to it. + static readonly string[] Collections = ["ProcessedMessages", "SagaSnapshots"]; + + public async Task ApplyRetention(TimeSpan retentionPeriod, IProgress progress, CancellationToken cancellationToken = default) + { + var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); + long updated = 0; + + foreach (var collection in Collections) + { + var query = new IndexQuery + { + Query = $$""" + from {{collection}} + update + { + this['@metadata']['@expires'] = new Date(new Date(this.ProcessedAt).getTime() + $retentionMilliseconds).toISOString(); + } + """, + QueryParameters = new Parameters + { + { "retentionMilliseconds", retentionPeriod.TotalMilliseconds } + } + }; + + var patch = new PatchByQueryOperation(query, new QueryOperationOptions { AllowStale = true }); + var operation = await documentStore.Operations.SendAsync(patch, token: cancellationToken); + + var updatedBefore = updated; + operation.OnProgressChanged += (_, operationProgress) => + { + if (operationProgress is DeterminateProgress determinate) + { + progress.Report(updatedBefore + determinate.Processed); + } + }; + + var result = await operation.WaitForCompletionAsync(cancellationToken); + updated += result.Total; + + logger.LogInformation("Updated the expiry of {Count:N0} {Collection} documents", result.Total, collection); + } + + return updated; + } + } +} diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistence.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistence.cs index 553014885a..ef832fb791 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistence.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistence.cs @@ -30,6 +30,7 @@ public void AddPersistence(IServiceCollection services) services.AddSingleton(sp => sp.GetRequiredService()); services.AddSingleton(); services.AddSingleton(); + services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); services.AddSingleton(); diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApplyRetentionTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApplyRetentionTests.cs new file mode 100644 index 0000000000..2da8a3b513 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApplyRetentionTests.cs @@ -0,0 +1,74 @@ +namespace ServiceControl.Audit.Persistence.Tests +{ + using System; + using System.Collections.Generic; + using System.Threading.Tasks; + using Auditing; + using Microsoft.Extensions.DependencyInjection; + using NServiceBus; + using NUnit.Framework; + using Raven.Client; + using SagaAudit; + + [TestFixture] + class ApplyRetentionTests : PersistenceTestFixture + { + public override Task Setup() + { + SetSettings = s => s.AuditRetentionPeriod = IngestionRetentionPeriod; + return base.Setup(); + } + + [Test] + public async Task Sets_the_expiry_of_stored_messages_to_processed_time_plus_the_new_retention_period() + { + var message = new ProcessedMessage( + new Dictionary + { + { Headers.MessageId, "MyMessageId" }, + { Headers.ProcessingEndpoint, "SomeEndpoint" }, + { Headers.ProcessingEnded, DateTimeOffsetHelper.ToWireFormattedString(new DateTimeOffset(ProcessedAt)) } + }, + new Dictionary { { "MessageId", "MyMessageId" } }); + var snapshot = new SagaSnapshot { SagaId = Guid.NewGuid(), ProcessedAt = ProcessedAt }; + + await using (var unitOfWork = await StartAuditUnitOfWork(2)) + { + await unitOfWork.RecordProcessedMessage(message); + await unitOfWork.RecordSagaSnapshot(snapshot); + await unitOfWork.Complete(); + } + await configuration.CompleteDBOperation(); + + var expiresBefore = await GetExpires(message.Id); + Assert.That(expiresBefore, Is.EqualTo(DateTime.UtcNow + IngestionRetentionPeriod).Within(TimeSpan.FromMinutes(5)), "ingestion sets the expiry from the configured retention period"); + + var progress = new List(); + var updated = await ServiceProvider.GetRequiredService() + .ApplyRetention(NewRetentionPeriod, new SynchronousProgress(progress), TestTimeoutCancellationToken); + + Assert.That(updated, Is.EqualTo(2)); + Assert.That(await GetExpires(message.Id), Is.EqualTo(ProcessedAt + NewRetentionPeriod)); + Assert.That(await GetExpires(snapshot.Id), Is.EqualTo(ProcessedAt + NewRetentionPeriod)); + } + + async Task GetExpires(string id) + { + using var session = configuration.DocumentStore.OpenAsyncSession(); + var document = await session.LoadAsync(id); + Assert.That(document, Is.Not.Null, $"document {id} not found"); + var expires = session.Advanced.GetMetadataFor(document).GetString(Constants.Documents.Metadata.Expires); + return DateTime.Parse(expires, null, System.Globalization.DateTimeStyles.AdjustToUniversal | System.Globalization.DateTimeStyles.AssumeUniversal); + } + + // Progress posts to the thread pool; this one reports inline so the test can read the list afterwards + sealed class SynchronousProgress(List reports) : IProgress + { + public void Report(long value) => reports.Add(value); + } + + static readonly DateTime ProcessedAt = new(2026, 1, 15, 10, 30, 0, DateTimeKind.Utc); + static readonly TimeSpan IngestionRetentionPeriod = TimeSpan.FromDays(30); + static readonly TimeSpan NewRetentionPeriod = TimeSpan.FromHours(5); + } +} diff --git a/src/ServiceControl.Audit.Persistence/IAuditRetentionApplier.cs b/src/ServiceControl.Audit.Persistence/IAuditRetentionApplier.cs new file mode 100644 index 0000000000..26dfcc789e --- /dev/null +++ b/src/ServiceControl.Audit.Persistence/IAuditRetentionApplier.cs @@ -0,0 +1,20 @@ +namespace ServiceControl.Audit.Persistence +{ + using System; + using System.Threading; + using System.Threading.Tasks; + + /// + /// Applies the configured audit retention period to the messages and saga snapshots that are already stored. + /// Only a persister that stores an expiry per document at ingestion registers an implementation. A persister + /// that applies the retention period when it removes expired messages does not need one. + /// + public interface IAuditRetentionApplier + { + /// + /// Sets the expiry of every stored message and saga snapshot to its processed time plus . + /// Returns the number of documents that were updated. + /// + Task ApplyRetention(TimeSpan retentionPeriod, IProgress progress, CancellationToken cancellationToken = default); + } +} diff --git a/src/ServiceControl.Audit.UnitTests/Infrastructure/HostArgumentsTests.cs b/src/ServiceControl.Audit.UnitTests/Infrastructure/HostArgumentsTests.cs new file mode 100644 index 0000000000..95ad0a845e --- /dev/null +++ b/src/ServiceControl.Audit.UnitTests/Infrastructure/HostArgumentsTests.cs @@ -0,0 +1,22 @@ +namespace ServiceControl.Audit.UnitTests.Infrastructure +{ + using Audit.Infrastructure.Hosting; + using Audit.Infrastructure.Hosting.Commands; + using NUnit.Framework; + + [TestFixture] + class HostArgumentsTests + { + [Test] + public void No_arguments_runs_the_instance() => + Assert.That(new HostArguments([]).Command, Is.EqualTo(typeof(RunCommand))); + + [Test] + public void Apply_retention_selects_the_apply_retention_command() => + Assert.That(new HostArguments(["--apply-retention"]).Command, Is.EqualTo(typeof(ApplyRetentionCommand))); + + [Test] + public void Import_failed_audits_selects_the_import_command() => + Assert.That(new HostArguments(["--import-failed-audits"]).Command, Is.EqualTo(typeof(ImportFailedAuditsCommand))); + } +} diff --git a/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/ApplyRetentionCommand.cs b/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/ApplyRetentionCommand.cs new file mode 100644 index 0000000000..b5fb635dde --- /dev/null +++ b/src/ServiceControl.Audit/Infrastructure/Hosting/Commands/ApplyRetentionCommand.cs @@ -0,0 +1,67 @@ +namespace ServiceControl.Audit.Infrastructure.Hosting.Commands +{ + using System; + using System.Threading; + using System.Threading.Tasks; + using Microsoft.Extensions.DependencyInjection; + using Microsoft.Extensions.Hosting; + using Microsoft.Extensions.Logging; + using Persistence; + using ServiceControl.Infrastructure; + using Settings; + + /// + /// Applies the configured audit retention period to the stored messages. Run it after a change of + /// ServiceControl.Audit/AuditRetentionPeriod: ingestion sets the expiry of each message when it is + /// stored, so the stored messages keep the old retention period until this command rewrites their expiry. + /// + class ApplyRetentionCommand : AbstractCommand + { + public override async Task Execute(HostArguments args, Settings settings, CancellationToken cancellationToken = default) + { + var logger = LoggerUtil.CreateStaticLogger(); + + var persistenceConfiguration = PersistenceConfigurationFactory.LoadPersistenceConfiguration(settings); + var persistenceSettings = persistenceConfiguration.BuildPersistenceSettings(settings); + + var hostBuilder = Host.CreateApplicationBuilder(); + hostBuilder.Services.AddPersistence(persistenceSettings, persistenceConfiguration); + + using var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + Console.CancelKeyPress += (_, e) => + { + e.Cancel = true; + tokenSource.Cancel(); + }; + + using var host = hostBuilder.Build(); + await host.StartAsync(tokenSource.Token); + + try + { + var applier = host.Services.GetService(); + + if (applier == null) + { + logger.LogInformation("The {PersistenceType} persister applies the retention period when it removes expired messages, so the stored messages need no update", settings.PersistenceType); + return; + } + + logger.LogInformation("Applying the audit retention period of {RetentionPeriod} to the stored messages", settings.AuditRetentionPeriod); + + var progress = new Progress(count => logger.LogInformation("{Count:N0} documents updated so far", count)); + var updated = await applier.ApplyRetention(settings.AuditRetentionPeriod, progress, tokenSource.Token); + + logger.LogInformation("Done. Updated the expiry of {Count:N0} documents", updated); + } + catch (OperationCanceledException e) when (tokenSource.Token.IsCancellationRequested) + { + logger.LogInformation(e, "Cancelled"); + } + finally + { + await host.StopAsync(CancellationToken.None); + } + } + } +} diff --git a/src/ServiceControl.Audit/Infrastructure/Hosting/Help.txt b/src/ServiceControl.Audit/Infrastructure/Hosting/Help.txt index 30fad00314..6ac1eb30b4 100644 --- a/src/ServiceControl.Audit/Infrastructure/Hosting/Help.txt +++ b/src/ServiceControl.Audit/Infrastructure/Hosting/Help.txt @@ -10,6 +10,19 @@ The REST API, message importers and the background document expiry are all disab This mode is only supported when run interactively. +APPLY RETENTION + + ServiceControl.Audit.exe --apply-retention + +Applies the configured audit retention period (ServiceControl.Audit/AuditRetentionPeriod) to the messages and saga +snapshots that are already stored. Ingestion sets the expiry of each message from the retention period that is +configured at that moment, so a change of the retention period only affects new messages until this command runs. +The new expiry of each message is its processed time plus the retention period. Messages that are already past the new +expiry are removed by the next expiration run. + +Stop the instance before running this command. A persister that applies the retention period when it removes expired +messages needs no update, and the command reports that. + SERVICE INSTALL AND UNINSTALL AND CONFIGURATION OPTIONS As of Service Control 1.7 the command line uninstall and install switches have been removed. diff --git a/src/ServiceControl.Audit/Infrastructure/Hosting/HostArguments.cs b/src/ServiceControl.Audit/Infrastructure/Hosting/HostArguments.cs index 5221dcdc0a..2a0ca08d90 100644 --- a/src/ServiceControl.Audit/Infrastructure/Hosting/HostArguments.cs +++ b/src/ServiceControl.Audit/Infrastructure/Hosting/HostArguments.cs @@ -53,6 +53,15 @@ public HostArguments(string[] args) } }; + var applyRetentionOptions = new OptionSet + { + { + "apply-retention", + "Apply the configured audit retention period to the stored messages", + s => Command = typeof(ApplyRetentionCommand) + } + }; + try { externalInstallerOptions.Parse(args); @@ -76,6 +85,13 @@ public HostArguments(string[] args) return; } + applyRetentionOptions.Parse(args); + + if (Command == typeof(ApplyRetentionCommand)) + { + return; + } + defaultOptions.Parse(args); } catch (Exception e)