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)