diff --git a/.github/workflows/build-db-container.yml b/.github/workflows/build-db-container.yml index 643f44ce30..c6b61cd4cb 100644 --- a/.github/workflows/build-db-container.yml +++ b/.github/workflows/build-db-container.yml @@ -55,7 +55,7 @@ jobs: $containers = cat src/ServiceControl.RavenDB/containers.json | ConvertFrom-Json $containers | ForEach-Object -Process { $FULLTAG="${{ env.TAG_NAME }}-$($_.tag)" - $BASETAG="$($RavenVersion)-ubuntu.22.04-$($_.tag)" + $BASETAG="$($RavenVersion)-ubuntu.24.04-$($_.tag)" Write-Output "::group::Building $FULLTAG for architecture $($_.arch) from $BASETAG" docker build -t ghcr.io/particular/servicecontrol-ravendb:$FULLTAG --file src/ServiceControl.RavenDB/Dockerfile --build-arg VERSION=${{ env.MinVerVersion }} --build-arg BASETAG=$BASETAG --platform linux/$($_.arch) . Write-Output "::endgroup::" diff --git a/docs/usage-report.md b/docs/usage-report.md index 3b56231f9a..0e110b9e89 100644 --- a/docs/usage-report.md +++ b/docs/usage-report.md @@ -89,7 +89,7 @@ Releases up to 6.21.2 report `SqlVersion` as the full text of `@@VERSION` on SQL | `Storage.BodyStorage.Type` | `RavenAttachments`, `FileSystem`, `AzureBlob`, `S3` | the persister | 6.20.0 | | `Limits.MaxBodySizeToStore` | bytes | `MaxBodySizeToStore`, SQL Server and PostgreSQL only | 6.20.0 | | `Storage.Schema` | `Default`, `Custom` | `Database/Schema`, SQL Server and PostgreSQL only | Unreleased | -| `Storage.LogLevel` | `None`, `Information`, `Operations` | `RavenDBLogLevel`, RavenDB only | Unreleased | +| `Storage.LogLevel` | `Trace`, `Debug`, `Info`, `Warn`, `Error`, `Fatal`, `Off` | `RavenDBLogLevel`, RavenDB only | Unreleased | | `Storage.CommandTimeoutSeconds` | `Default` or number | `Database/CommandTimeout`, SQL Server and PostgreSQL only | Unreleased | | `Storage.QueryTimeoutSeconds` | `Default` or number | `QueryTimeoutInSeconds` | Unreleased | | `Storage.SubscriptionCacheSeconds` | `Default` or number | `SubscriptionCacheDuration`, SQL Server and PostgreSQL only | Unreleased | diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props index 2860b108d4..8469bc4f51 100644 --- a/src/Directory.Packages.props +++ b/src/Directory.Packages.props @@ -79,7 +79,7 @@ - + diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs index 99c74f6a15..104e6da960 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs @@ -39,7 +39,7 @@ public async Task Initialize(CancellationToken cancellationToken = default) var serverConfig = databaseConfiguration.ServerConfiguration; - var embeddedConfig = new EmbeddedDatabaseConfiguration(serverConfig.ServerUrl, databaseConfiguration.Name, serverConfig.DbPath, serverConfig.LogPath, serverConfig.LogsMode); + var embeddedConfig = new EmbeddedDatabaseConfiguration(serverConfig.ServerUrl, databaseConfiguration.Name, serverConfig.DbPath, serverConfig.LogPath, serverConfig.LogsMinLevel); database = EmbeddedDatabase.Start(embeddedConfig, lifetime); diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs index 316128e68e..0dbbce5e7e 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs @@ -5,11 +5,12 @@ namespace ServiceControl.Audit.Persistence.RavenDB using System; using System.Threading; using System.Threading.Tasks; + using Microsoft.Extensions.Logging; using Raven.Client.Documents; using Raven.Client.Documents.Conventions; using ServiceControl.RavenDB; - sealed class RavenExternalPersistenceLifecycle(DatabaseConfiguration configuration) : IRavenPersistenceLifecycle, IRavenDocumentStoreProvider, IDisposable + sealed class RavenExternalPersistenceLifecycle(DatabaseConfiguration configuration, ILogger logger) : IRavenPersistenceLifecycle, IRavenDocumentStoreProvider, IDisposable { public async ValueTask GetDocumentStore(CancellationToken cancellationToken = default) { @@ -42,7 +43,8 @@ public async Task Initialize(CancellationToken cancellationToken = default) Certificate = RavenClientCertificate.FindClientCertificate(configuration.ServerConfiguration), Conventions = new DocumentConventions { - SaveEnumsAsIntegers = true + SaveEnumsAsIntegers = true, + SaveEnumsAsIntegersForPatching = true } }; @@ -51,7 +53,7 @@ public async Task Initialize(CancellationToken cancellationToken = default) await StartupChecks.EnsureServerVersion(store, cancellationToken); var databaseSetup = new DatabaseSetup(configuration); - await databaseSetup.Execute(store, cancellationToken); + await DatabaseLoadRetry.Execute(token => databaseSetup.Execute(store, token), logger, cancellationToken); } finally { diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs index 787a06d9c1..68ffed5c3d 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenPersistenceConfiguration.cs @@ -109,14 +109,14 @@ internal static DatabaseConfiguration GetDatabaseConfiguration(PersistenceSettin var logPath = GetLogPath(settings); - var logsMode = "Operations"; + var logsMinLevel = "Warn"; if (settings.PersisterSpecificSettings.TryGetValue(RavenDbLogLevelKey, out var ravenDbLogLevel)) { - logsMode = RavenDbLogLevelToLogsModeMapper.Map(ravenDbLogLevel, Logger); + logsMinLevel = RavenDbLogLevelToLogsMinLevelMapper.Map(ravenDbLogLevel, Logger); } - serverConfiguration = new ServerConfiguration(dbPath, serverUrl, logPath, logsMode); + serverConfiguration = new ServerConfiguration(dbPath, serverUrl, logPath, logsMinLevel); } var dataSpaceRemainingThreshold = CheckFreeDiskSpace.Parse(settings.PersisterSpecificSettings, Logger); diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/ServerConfiguration.cs b/src/ServiceControl.Audit.Persistence.RavenDB/ServerConfiguration.cs index 53c3199f31..fc60106427 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/ServerConfiguration.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/ServerConfiguration.cs @@ -10,13 +10,13 @@ public ServerConfiguration(string connectionString) ConnectionString = connectionString; } - public ServerConfiguration(string dbPath, string serverUrl, string logPath, string logsMode) + public ServerConfiguration(string dbPath, string serverUrl, string logPath, string logsMinLevel) { UseEmbeddedServer = true; DbPath = dbPath; ServerUrl = serverUrl; LogPath = logPath; - LogsMode = logsMode; + LogsMinLevel = logsMinLevel; } public string ConnectionString { get; } @@ -27,6 +27,6 @@ public ServerConfiguration(string dbPath, string serverUrl, string logPath, stri public string DbPath { get; internal set; } //Setter for ATT only public string ServerUrl { get; } public string LogPath { get; } - public string LogsMode { get; set; } + public string LogsMinLevel { get; set; } } } diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApprovalFiles/IndexDefinitionTests.VerifyGeneratedDefinitions.approved.txt b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApprovalFiles/IndexDefinitionTests.VerifyGeneratedDefinitions.approved.txt new file mode 100644 index 0000000000..d14ec4f9cd --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ApprovalFiles/IndexDefinitionTests.VerifyGeneratedDefinitions.approved.txt @@ -0,0 +1,121 @@ +Index: FailedAuditImportIndex +Type: Map +SourceType: Documents +Maps: + docs.FailedAuditImports.Select(cc => new { + Id = Id(cc) +}) +Reduce: (none) +Fields: +Configuration: + Indexing.Static.SearchEngineType=Lucene +AdditionalSources: +AdditionalAssemblies: 0 +OutputReduceToCollection: (none) + +Index: MessagesViewIndex +Type: Map +SourceType: Documents +Maps: + docs.ProcessedMessages.Select(message => new { + MessageId = ((string) message.MessageMetadata["MessageId"]), + MessageType = ((string) message.MessageMetadata["MessageType"]), + IsSystemMessage = ((bool) message.MessageMetadata["IsSystemMessage"]), + Status = ((bool) message.MessageMetadata["IsRetried"]) ? 4 : 3, + TimeSent = ((DateTime) message.MessageMetadata["TimeSent"]), + ProcessedAt = message.ProcessedAt, + ReceivingEndpointName = ((message.MessageMetadata["ReceivingEndpoint"])).Name, + SendingEndpointName = ((message.MessageMetadata["SendingEndpoint"])).Name, + CriticalTime = ((TimeSpan ? ) message.MessageMetadata["CriticalTime"]), + ProcessingTime = ((TimeSpan ? ) message.MessageMetadata["ProcessingTime"]), + DeliveryTime = ((TimeSpan ? ) message.MessageMetadata["DeliveryTime"]), + Query = Enumerable.ToArray(DynamicEnumerable.Union(message.MessageMetadata.ToDictionary(e1 => e1.Key, e1 => e1.Value).Select(_ => _.Value.ToString()), new string[] { + String.Join(" ", message.Headers.ToDictionary(e1 => e1.Key, e1 => e1.Value).Select(x => x.Value)) + })), + ConversationId = ((string) message.MessageMetadata["ConversationId"]) +}) +Reduce: (none) +Fields: + Query: Storage=, Indexing=Search, TermVector=, Suggestions=, Analyzer=Lucene.Net.Analysis.Standard.StandardAnalyzer, Lucene.Net, Version=3.0.3.0, Culture=neutral, PublicKeyToken=85089178b9ac3181, HasSpatial=False +Configuration: + Indexing.Static.SearchEngineType=Lucene +AdditionalSources: +AdditionalAssemblies: 0 +OutputReduceToCollection: (none) + +Index: MessagesViewIndexWithFullTextSearch +Type: Map +SourceType: Documents +Maps: + docs.ProcessedMessages.Select(message => new { + MessageId = ((string) message.MessageMetadata["MessageId"]), + MessageType = ((string) message.MessageMetadata["MessageType"]), + IsSystemMessage = ((bool) message.MessageMetadata["IsSystemMessage"]), + Status = ((bool) message.MessageMetadata["IsRetried"]) ? 4 : 3, + TimeSent = ((DateTime) message.MessageMetadata["TimeSent"]), + ProcessedAt = message.ProcessedAt, + ReceivingEndpointName = ((message.MessageMetadata["ReceivingEndpoint"])).Name, + SendingEndpointName = ((message.MessageMetadata["SendingEndpoint"])).Name, + CriticalTime = ((TimeSpan ? ) message.MessageMetadata["CriticalTime"]), + ProcessingTime = ((TimeSpan ? ) message.MessageMetadata["ProcessingTime"]), + DeliveryTime = ((TimeSpan ? ) message.MessageMetadata["DeliveryTime"]), + Query = Enumerable.ToArray(DynamicEnumerable.Union(message.MessageMetadata.ToDictionary(e1 => e1.Key, e1 => e1.Value).Select(_ => _.Value.ToString()), new string[] { + String.Join(" ", message.Headers.ToDictionary(e1 => e1.Key, e1 => e1.Value).Select(x => x.Value)), + this.LoadAttachment(message, "body").GetContentAsString() + })), + ConversationId = ((string) message.MessageMetadata["ConversationId"]) +}) +Reduce: (none) +Fields: + Query: Storage=, Indexing=Search, TermVector=, Suggestions=, Analyzer=Lucene.Net.Analysis.Standard.StandardAnalyzer, Lucene.Net, Version=3.0.3.0, Culture=neutral, PublicKeyToken=85089178b9ac3181, HasSpatial=False +Configuration: + Indexing.Static.SearchEngineType=Lucene +AdditionalSources: +AdditionalAssemblies: 0 +OutputReduceToCollection: (none) + +Index: SagaDetailsIndex +Type: MapReduce +SourceType: Documents +Maps: + docs.SagaHistories.Select(doc => new { + SagaId = doc.SagaId, + Id = doc.SagaId, + SagaType = doc.SagaType, + Changes = doc.Changes +}) + docs.SagaSnapshots.Select(doc => new { + SagaId = doc.SagaId, + Id = doc.SagaId, + SagaType = doc.SagaType, + Changes = new[] { + new { + Endpoint = doc.Endpoint, + FinishTime = doc.FinishTime, + InitiatingMessage = doc.InitiatingMessage, + OutgoingMessages = doc.OutgoingMessages, + StartTime = doc.StartTime, + StateAfterChange = doc.StateAfterChange, + Status = doc.Status + } + } +}) +Reduce: results.GroupBy(result => result.SagaId).Select(g => new { + g = g, + first = DynamicEnumerable.FirstOrDefault(g) +}).Select(this0 => new { + Id = this0.first.SagaId, + SagaId = this0.first.SagaId, + SagaType = this0.first.SagaType, + Changes = Enumerable.ToList(DynamicEnumerable.OrderByDescending(this0.g.SelectMany(x => x.Changes), x0 => x0.FinishTime).Take(50000)) +}) +Fields: + Changes: Storage=, Indexing=No, TermVector=, Suggestions=, Analyzer=(none), HasSpatial=False + SagaId: Storage=, Indexing=Exact, TermVector=, Suggestions=, Analyzer=(none), HasSpatial=False + SagaType: Storage=, Indexing=No, TermVector=, Suggestions=, Analyzer=(none), HasSpatial=False +Configuration: + Indexing.Static.SearchEngineType=Lucene +AdditionalSources: +AdditionalAssemblies: 0 +OutputReduceToCollection: (none) + diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/AuditTestData.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/AuditTestData.cs new file mode 100644 index 0000000000..1260adfb16 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/AuditTestData.cs @@ -0,0 +1,57 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System; +using System.Collections.Generic; +using Auditing; +using Monitoring; +using NServiceBus; + +static class AuditTestData +{ + // Carries every metadata entry that the audit indexes read, as the audit ingestion does. + public static ProcessedMessage CreateProcessedMessage( + string messageId = null, + string messageType = "MyMessageType", + string conversationId = null, + string receivingEndpoint = "ReceivingEndpoint", + DateTime? timeSent = null, + TimeSpan? criticalTime = null, + TimeSpan? processingTime = null, + TimeSpan? deliveryTime = null, + DateTimeOffset? processedAt = null) + { + messageId ??= Guid.NewGuid().ToString(); + conversationId ??= Guid.NewGuid().ToString(); + timeSent ??= DateTime.UtcNow; + processedAt ??= DateTimeOffset.UtcNow; + + var metadata = new Dictionary + { + { "MessageId", messageId }, + { "MessageIntent", MessageIntent.Send }, + { "TimeSent", timeSent.Value }, + { "CriticalTime", criticalTime ?? TimeSpan.FromSeconds(5) }, + { "ProcessingTime", processingTime ?? TimeSpan.FromSeconds(1) }, + { "DeliveryTime", deliveryTime ?? TimeSpan.FromSeconds(4) }, + { "IsSystemMessage", false }, + { "MessageType", messageType }, + { "IsRetried", false }, + { "ConversationId", conversationId }, + { "SendingEndpoint", new EndpointDetails { Name = "SendingEndpoint", HostId = Guid.Empty, Host = "host" } }, + { "ReceivingEndpoint", new EndpointDetails { Name = receivingEndpoint, HostId = Guid.Empty, Host = "host" } } + }; + + var headers = new Dictionary + { + { Headers.MessageId, messageId }, + { Headers.ProcessingEndpoint, receivingEndpoint }, + { Headers.MessageIntent, MessageIntent.Send.ToString() }, + { Headers.ConversationId, conversationId }, + { Headers.ProcessingStarted, DateTimeOffsetHelper.ToWireFormattedString(processedAt.Value) }, + { Headers.ProcessingEnded, DateTimeOffsetHelper.ToWireFormattedString(processedAt.Value) }, + { Headers.EnclosedMessageTypes, messageType } + }; + + return new ProcessedMessage(headers, metadata); + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/CleanDatabaseFixture.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/CleanDatabaseFixture.cs new file mode 100644 index 0000000000..b209e7f4ab --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/CleanDatabaseFixture.cs @@ -0,0 +1,55 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; +using Raven.Client.Documents; +using Raven.Client.Documents.Conventions; +using Raven.Client.ServerWide; +using Raven.Client.ServerWide.Operations; + +// A new database on the shared embedded server with no settings and no indexes, as opposed to the one DatabaseSetup prepares. +abstract class CleanDatabaseFixture +{ + protected IDocumentStore Store { get; private set; } + + protected CancellationToken TestTimeoutCancellationToken => testCancellationTokenSource.Token; + + [SetUp] + public async Task CreateCleanDatabase() + { + testCancellationTokenSource = new CancellationTokenSource(TestTimeout); + + var embeddedDatabase = await SharedEmbeddedServer.GetInstance(TestContext.CurrentContext.CancellationToken); + databaseName = Guid.NewGuid().ToString(); + + Store = new DocumentStore + { + Urls = [embeddedDatabase.ServerUrl], + Database = databaseName, + Conventions = new DocumentConventions { SaveEnumsAsIntegers = true, SaveEnumsAsIntegersForPatching = true } + }; + Store.Initialize(); + + await Store.Maintenance.Server.SendAsync(new CreateDatabaseOperation(new DatabaseRecord(databaseName)), TestTimeoutCancellationToken); + } + + [TearDown] + public async Task DeleteCleanDatabase() + { + if (Store != null) + { + await Store.Maintenance.Server.SendAsync(new DeleteDatabasesOperation( + new DeleteDatabasesOperation.Parameters { DatabaseNames = [databaseName], HardDelete = true })); + Store.Dispose(); + } + + testCancellationTokenSource?.Dispose(); + } + + string databaseName; + CancellationTokenSource testCancellationTokenSource; + + static readonly TimeSpan TestTimeout = TimeSpan.FromSeconds(60); +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ExpirationMetadataTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ExpirationMetadataTests.cs new file mode 100644 index 0000000000..a098a44c5e --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/ExpirationMetadataTests.cs @@ -0,0 +1,71 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.Globalization; +using System.Text; +using System.Threading.Tasks; +using Auditing; +using NUnit.Framework; +using Raven.Client; +using SagaAudit; + +[TestFixture] +class ExpirationMetadataTests : PersistenceTestFixture +{ + static readonly TimeSpan RetentionPeriod = TimeSpan.FromDays(7); + + public override Task Setup() + { + SetSettings = s => s.AuditRetentionPeriod = RetentionPeriod; + return base.Setup(); + } + + [Test] + public async Task Processed_message_should_expire_with_its_body_after_the_retention_period() + { + var message = AuditTestData.CreateProcessedMessage(); + + await using (var unitOfWork = await StartAuditUnitOfWork(1)) + { + await unitOfWork.RecordProcessedMessage(message, Encoding.UTF8.GetBytes("The body")); + await unitOfWork.Complete(); + } + + using var session = configuration.DocumentStore.OpenAsyncSession(); + var document = await session.LoadAsync(message.Id, TestContext.CurrentContext.CancellationToken); + var metadata = session.Advanced.GetMetadataFor(document); + + using (Assert.EnterMultipleScope()) + { + AssertExpiresAfterRetentionPeriod(metadata); + Assert.That(await session.Advanced.Attachments.ExistsAsync(message.Id, "body", TestContext.CurrentContext.CancellationToken), Is.True, "The body is an attachment of the message document, so it expires with it"); + } + } + + [Test] + public async Task Saga_snapshot_should_expire_after_the_retention_period() + { + var snapshot = new SagaSnapshot { SagaId = Guid.NewGuid() }; + + await using (var unitOfWork = await StartAuditUnitOfWork(1)) + { + await unitOfWork.RecordSagaSnapshot(snapshot); + await unitOfWork.Complete(); + } + + using var session = configuration.DocumentStore.OpenAsyncSession(); + var document = await session.LoadAsync(snapshot.Id, TestContext.CurrentContext.CancellationToken); + + AssertExpiresAfterRetentionPeriod(session.Advanced.GetMetadataFor(document)); + } + + static void AssertExpiresAfterRetentionPeriod(IDictionary metadata) + { + Assert.That(metadata.TryGetValue(Constants.Documents.Metadata.Expires, out var expiresValue), Is.True, $"{Constants.Documents.Metadata.Expires} is missing from the document metadata"); + + var expires = DateTime.Parse((string)expiresValue, null, DateTimeStyles.RoundtripKind); + + Assert.That(expires, Is.EqualTo(DateTime.UtcNow.Add(RetentionPeriod)).Within(TimeSpan.FromMinutes(5))); + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/IndexDefinitionTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/IndexDefinitionTests.cs new file mode 100644 index 0000000000..4441582926 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/IndexDefinitionTests.cs @@ -0,0 +1,79 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.Approvals; +using Persistence.RavenDB.Indexes; +using Raven.Client.Documents.Indexes; +using ServiceControl.RavenDB; +using ServiceControl.SagaAudit; + +[TestFixture] +class IndexDefinitionTests : CleanDatabaseFixture +{ + // A definition that differs from the one on the server makes RavenDB rebuild the index side by side, which takes + // days for the full-text index of a large audit database. Do not approve a change without a plan for the rebuild. + [Test] + public async Task VerifyGeneratedDefinitions() + { + AbstractIndexCreationTask[] indexes = + [ + new FailedAuditImportIndex(), + new MessagesViewIndex(), + new MessagesViewIndexWithFullTextSearch(), + new SagaDetailsIndex() + ]; + + await IndexDeployment.CreateIndexesAsync(indexes, Store, TestTimeoutCancellationToken); + + var description = new StringBuilder(); + + foreach (var index in indexes.OrderBy(i => i.IndexName, StringComparer.Ordinal)) + { + index.Conventions = Store.Conventions; + Describe(description, index.IndexName, index.CreateIndexDefinition()); + } + + Approver.Verify(description.ToString().ReplaceLineEndings()); + } + + static void Describe(StringBuilder description, string indexName, IndexDefinition definition) + { + description.AppendLine($"Index: {indexName}"); + description.AppendLine($"Type: {definition.Type}"); + description.AppendLine($"SourceType: {definition.SourceType}"); + + description.AppendLine("Maps:"); + foreach (var map in definition.Maps.OrderBy(m => m, StringComparer.Ordinal)) + { + description.AppendLine($" {map}"); + } + + description.AppendLine($"Reduce: {definition.Reduce ?? "(none)"}"); + + description.AppendLine("Fields:"); + foreach (var (name, options) in definition.Fields.OrderBy(f => f.Key, StringComparer.Ordinal)) + { + description.AppendLine($" {name}: Storage={options.Storage}, Indexing={options.Indexing}, TermVector={options.TermVector}, Suggestions={options.Suggestions}, Analyzer={options.Analyzer ?? "(none)"}, HasSpatial={options.Spatial != null}"); + } + + description.AppendLine("Configuration:"); + foreach (var (key, value) in definition.Configuration.OrderBy(c => c.Key, StringComparer.Ordinal)) + { + description.AppendLine($" {key}={value}"); + } + + description.AppendLine("AdditionalSources:"); + foreach (var (name, source) in definition.AdditionalSources.OrderBy(s => s.Key, StringComparer.Ordinal)) + { + description.AppendLine($" {name}: {source}"); + } + + description.AppendLine($"AdditionalAssemblies: {definition.AdditionalAssemblies.Count}"); + description.AppendLine($"OutputReduceToCollection: {definition.OutputReduceToCollection ?? "(none)"}"); + description.AppendLine(); + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/IndexDeploymentTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/IndexDeploymentTests.cs new file mode 100644 index 0000000000..4943a3fefd --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/IndexDeploymentTests.cs @@ -0,0 +1,150 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System; +using System.Collections.Generic; +using System.IO; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using Auditing; +using NUnit.Framework; +using Persistence.RavenDB.Indexes; +using Raven.Client.Documents; +using Raven.Client.Documents.Indexes; +using Raven.Client.Documents.Operations.Indexes; +using Raven.Client.Documents.Session; +using ServiceControl.RavenDB; +using ServiceControl.SagaAudit; + +[TestFixture] +class IndexDeploymentTests : CleanDatabaseFixture +{ + const string BodyText = "The zebrafish swims upstream"; + const string BodySearchTerm = "zebrafish"; + const string AnalyzerTypeName = "Lucene.Net.Analysis.Standard.StandardAnalyzer, Lucene.Net, Version=3.0.3.0, Culture=neutral, PublicKeyToken=85089178b9ac3181"; + + [TestCase(typeof(MessagesViewIndexWithFullTextSearch))] + [TestCase(typeof(MessagesViewIndex))] + [TestCase(typeof(SagaDetailsIndex))] + [TestCase(typeof(FailedAuditImportIndex))] + public async Task Index_should_deploy_to_a_clean_database_and_be_queryable(Type indexType) + { + var index = (AbstractIndexCreationTask)Activator.CreateInstance(indexType); + + await IndexDeployment.CreateIndexesAsync([index], Store, TestTimeoutCancellationToken); + + var definition = await Store.Maintenance.SendAsync(new GetIndexOperation(index.IndexName), TestTimeoutCancellationToken); + Assert.That(definition, Is.Not.Null); + + if (index is MessagesViewIndexWithFullTextSearch) + { + await AssertMessagesAreSearchable(index.IndexName, bodyIsSearchable: true); + } + else if (index is MessagesViewIndex) + { + await AssertMessagesAreSearchable(index.IndexName, bodyIsSearchable: false); + } + else if (index is SagaDetailsIndex) + { + await AssertSagaHistoryIsReduced(); + } + else + { + await AssertFailedAuditImportIsIndexed(); + } + + var stats = await Store.Maintenance.SendAsync(new GetIndexStatisticsOperation(index.IndexName), TestTimeoutCancellationToken); + + using (Assert.EnterMultipleScope()) + { + Assert.That(stats.State, Is.EqualTo(IndexState.Normal)); + Assert.That(stats.ErrorsCount, Is.Zero); + Assert.That(stats.SearchEngineType, Is.EqualTo(SearchEngineType.Lucene)); + } + } + + [TestCase(typeof(MessagesViewIndexWithFullTextSearch))] + [TestCase(typeof(MessagesViewIndex))] + public async Task Full_text_indexes_should_use_the_lucene_standard_analyzer_on_the_query_field(Type indexType) + { + var index = (AbstractIndexCreationTask)Activator.CreateInstance(indexType); + + await IndexDeployment.CreateIndexesAsync([index], Store, TestTimeoutCancellationToken); + + var definition = await Store.Maintenance.SendAsync(new GetIndexOperation(index.IndexName), TestTimeoutCancellationToken); + + Assert.That(definition.Fields[nameof(MessagesViewIndex.SortAndFilterOptions.Query)].Analyzer, Is.EqualTo(AnalyzerTypeName)); + } + + async Task AssertMessagesAreSearchable(string indexName, bool bodyIsSearchable) + { + var message = AuditTestData.CreateProcessedMessage(messageId: "MyMessageId"); + message.Id = $"ProcessedMessages-{DateTime.UtcNow.Ticks}-{message.GetProcessingId()}"; + + using (var session = Store.OpenAsyncSession()) + { + await session.StoreAsync(message, message.Id, TestTimeoutCancellationToken); + session.Advanced.Attachments.Store(message.Id, "body", new MemoryStream(Encoding.UTF8.GetBytes(BodyText)), "text/plain"); + await session.SaveChangesAsync(TestTimeoutCancellationToken); + } + + await Store.WaitForIndexingAsync(TestTimeoutCancellationToken); + + using (var session = Store.OpenAsyncSession()) + { + var byMetadata = await Search(session, indexName, "MyMessageId"); + var byBody = await Search(session, indexName, BodySearchTerm); + + using (Assert.EnterMultipleScope()) + { + Assert.That(byMetadata, Has.Count.EqualTo(1), "A search over the message metadata should find the message"); + Assert.That(byBody, Has.Count.EqualTo(bodyIsSearchable ? 1 : 0), "A search over the stored body should find the message only with the full-text index"); + } + } + + static Task> Search(IAsyncDocumentSession session, string indexName, string searchTerm) => + session.Query(indexName) + .Search(x => x.Query, searchTerm) + .ToListAsync(); + } + + async Task AssertSagaHistoryIsReduced() + { + var sagaId = Guid.NewGuid(); + + using (var session = Store.OpenAsyncSession()) + { + await session.StoreAsync(new SagaSnapshot { SagaId = sagaId, SagaType = "MySagaType", FinishTime = DateTime.UtcNow }, TestTimeoutCancellationToken); + await session.StoreAsync(new SagaSnapshot { SagaId = sagaId, SagaType = "MySagaType", FinishTime = DateTime.UtcNow.AddSeconds(1) }, TestTimeoutCancellationToken); + await session.SaveChangesAsync(TestTimeoutCancellationToken); + } + + await Store.WaitForIndexingAsync(TestTimeoutCancellationToken); + + using (var session = Store.OpenAsyncSession()) + { + var history = await session.Query().SingleOrDefaultAsync(x => x.SagaId == sagaId, TestTimeoutCancellationToken); + + Assert.That(history, Is.Not.Null); + Assert.That(history.Changes, Has.Count.EqualTo(2)); + } + } + + async Task AssertFailedAuditImportIsIndexed() + { + using (var session = Store.OpenAsyncSession()) + { + await session.StoreAsync(new FailedAuditImport(), "FailedAuditImports/1", TestTimeoutCancellationToken); + await session.SaveChangesAsync(TestTimeoutCancellationToken); + } + + await Store.WaitForIndexingAsync(TestTimeoutCancellationToken); + + using (var session = Store.OpenAsyncSession()) + { + var count = await session.Query().CountAsync(TestTimeoutCancellationToken); + + Assert.That(count, Is.EqualTo(1)); + } + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/LicenseTest.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/LicenseTest.cs index 62d456afa0..dc8bb4c437 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/LicenseTest.cs +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/LicenseTest.cs @@ -4,8 +4,10 @@ using System.Linq; using System.Net.Http; using System.Text; + using System.Text.Json; using System.Threading.Tasks; using NUnit.Framework; + using Persistence.RavenDB; [NonParallelizable] class LicenseTest : PersistenceTestFixture @@ -44,6 +46,38 @@ public async Task EnsureLicenseIsValid() } } + // LicenseStatusCheck deserializes the response into its own type. A renamed property leaves the default value in place, + // so the check waits for the timeout or reports no problem, which is why the names are asserted on the raw response. + [Test] + public async Task License_status_should_contain_the_properties_read_by_the_license_check() + { + var ravenUrl = configuration.DocumentStore.Urls.First(); + + using var http = new HttpClient(); + var jsonText = await http.GetStringAsync(ravenUrl + "/license/status", TestTimeoutCancellationToken); + + using var document = JsonDocument.Parse(jsonText); + var license = document.RootElement; + + using (Assert.EnterMultipleScope()) + { + Assert.That(license.TryGetProperty("Id", out var id), Is.True, "Id is missing from the response"); + Assert.That(id.ValueKind, Is.EqualTo(JsonValueKind.String)); + Assert.That(license.TryGetProperty("LicensedTo", out var licensedTo), Is.True, "LicensedTo is missing from the response"); + Assert.That(licensedTo.ValueKind, Is.EqualTo(JsonValueKind.String)); + Assert.That(license.TryGetProperty("Status", out var status), Is.True, "Status is missing from the response"); + Assert.That(status.ValueKind, Is.EqualTo(JsonValueKind.String)); + Assert.That(license.TryGetProperty("Expired", out var expired), Is.True, "Expired is missing from the response"); + Assert.That(expired.ValueKind, Is.AnyOf(JsonValueKind.True, JsonValueKind.False)); + } + } + + [Test] + public async Task License_status_check_should_accept_the_license_of_the_server() + { + await LicenseStatusCheck.WaitForLicenseOrThrow(configuration.DocumentStore, TestTimeoutCancellationToken); + } + class LicenseDetails { public string Id { get; set; } diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/MemoryInformationRetrieverTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/MemoryInformationRetrieverTests.cs new file mode 100644 index 0000000000..bcaed33521 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/MemoryInformationRetrieverTests.cs @@ -0,0 +1,45 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System.Linq; +using System.Net.Http; +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using Persistence.RavenDB; + +[TestFixture] +class MemoryInformationRetrieverTests : PersistenceTestFixture +{ + // MemoryInformationRetriever deserializes the response into its own types. A renamed property leaves the default + // value in place and the dirty memory check reports no problem, so the names are asserted on the raw response. + [Test] + public async Task Memory_stats_should_contain_the_properties_read_by_the_dirty_memory_check() + { + var requestUrl = configuration.DocumentStore.Urls.First().TrimEnd('/') + "/admin/debug/memory/stats?includeThreads=false&includeMappings=false"; + + using var http = new HttpClient(); + var json = await http.GetStringAsync(requestUrl, TestTimeoutCancellationToken); + + using var document = JsonDocument.Parse(json); + Assert.That(document.RootElement.TryGetProperty("MemoryInformation", out var memoryInformation), Is.True, "MemoryInformation is missing from the response"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(memoryInformation.TryGetProperty("IsHighDirty", out var isHighDirty), Is.True, "MemoryInformation.IsHighDirty is missing from the response"); + Assert.That(isHighDirty.ValueKind, Is.AnyOf(JsonValueKind.True, JsonValueKind.False)); + Assert.That(memoryInformation.TryGetProperty("DirtyMemory", out var dirtyMemory), Is.True, "MemoryInformation.DirtyMemory is missing from the response"); + Assert.That(dirtyMemory.ValueKind, Is.EqualTo(JsonValueKind.String)); + } + } + + [Test] + public async Task Should_retrieve_the_dirty_memory_of_the_server() + { + var retriever = ServiceProvider.GetRequiredService(); + + var (_, dirtyMemory) = await retriever.GetMemoryInformation(TestTimeoutCancellationToken); + + Assert.That(dirtyMemory, Is.Not.Null.And.Not.Empty); + } +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs index 7bdc6f00a3..99eb8718c9 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs @@ -2,6 +2,7 @@ { using System; using System.Collections.Generic; + using System.Text; using System.Threading; using System.Threading.Tasks; using Auditing; @@ -92,6 +93,32 @@ Task> QuerySagaHistory(CancellationToken cancellationTo SagaHistoryStore.QuerySagaHistoryById(sagaId, cancellationToken); } + [Test] + public async Task MessageBodyRetention() + { + await WaitForIndexesToBeBuilt(); + + var message = MakeMessage("MyMessageId"); + + await using (var unitOfWork = await StartAuditUnitOfWork(1)) + { + await unitOfWork.RecordProcessedMessage(message, Encoding.UTF8.GetBytes("The body")); + await unitOfWork.Complete(); + } + + Assert.That(await BodyExists(message.Id, TestContext.CurrentContext.CancellationToken), Is.True); + + var bodyExistsAfterExpiration = await WaitUntil(token => BodyExists(message.Id, token), exists => !exists); + + Assert.That(bodyExistsAfterExpiration, Is.False); + + async Task BodyExists(string documentId, CancellationToken cancellationToken) + { + using var session = configuration.DocumentStore.OpenAsyncSession(); + return await session.Advanced.Attachments.ExistsAsync(documentId, "body", cancellationToken); + } + } + // The retention window starts at ingestion, so building the indexes of the fresh database must // not happen inside it: on a busy runner that alone has taken longer than the window. Task WaitForIndexesToBeBuilt() => configuration.CompleteDBOperation(); diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SearchEngineQueryTests.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SearchEngineQueryTests.cs new file mode 100644 index 0000000000..b08808c2d2 --- /dev/null +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SearchEngineQueryTests.cs @@ -0,0 +1,225 @@ +namespace ServiceControl.Audit.Persistence.Tests; + +using System; +using System.Linq; +using System.Text; +using System.Threading.Tasks; +using Auditing.MessagesView; +using NUnit.Framework; +using Persistence.RavenDB; +using Persistence.RavenDB.Indexes; +using Raven.Client.Documents.Indexes; +using Raven.Client.Documents.Operations.Indexes; +using Raven.Client.ServerWide; +using Raven.Client.ServerWide.Operations; +using ServiceControl.Audit.Infrastructure; +using ServiceControl.RavenDB; +using ServiceControl.SagaAudit; + +// Databases created before ServiceControl pinned the engine have no search engine setting, and DatabaseSetup pins them to Corax. +// The Corax fixture reproduces that: the database and its indexes exist, pinned to Corax, before the persistence starts. +[TestFixture(SearchEngineType.Lucene)] +[TestFixture(SearchEngineType.Corax)] +class SearchEngineQueryTests(SearchEngineType searchEngine) : PersistenceTestFixture +{ + static readonly Guid SagaId = Guid.NewGuid(); + static readonly DateTime FirstTimeSent = DateTime.UtcNow.AddMinutes(-10); + + public override async Task Setup() + { + databaseName = Guid.NewGuid().ToString(); + + if (searchEngine == SearchEngineType.Corax) + { + await CreateExistingDatabaseWithCoraxIndexes(); + } + + SetSettings = s => s.PersisterSpecificSettings[RavenPersistenceConfiguration.DatabaseNameKey] = databaseName; + + await base.Setup(); + + await IngestAudits(); + } + + [Test] + public async Task Database_and_indexes_should_use_the_search_engine_under_test() + { + var databaseRecord = await configuration.DocumentStore.Maintenance.Server.SendAsync(new GetDatabaseRecordOperation(databaseName), TestTimeoutCancellationToken); + var indexStats = await configuration.DocumentStore.Maintenance.SendAsync(new GetIndexesStatisticsOperation(), TestTimeoutCancellationToken); + + using (Assert.EnterMultipleScope()) + { + Assert.That(databaseRecord.Settings["Indexing.Static.SearchEngineType"], Is.EqualTo(searchEngine.ToString())); + Assert.That(indexStats, Is.Not.Empty); + Assert.That(indexStats.Select(i => i.SearchEngineType), Has.All.EqualTo(searchEngine)); + } + } + + [Test] + public async Task Message_list_should_be_sorted_by_every_sort_option() + { + (string Sort, string Direction, string ExpectedOrder)[] sortOptions = + [ + ("time_sent", "asc", "A,B,C"), + ("time_sent", "desc", "C,B,A"), + ("message_id", "asc", "A,B,C"), + ("message_id", "desc", "C,B,A"), + ("message_type", "asc", "A,B,C"), + ("message_type", "desc", "C,B,A"), + ("critical_time", "asc", "B,C,A"), + ("critical_time", "desc", "A,C,B"), + ("processing_time", "asc", "C,A,B"), + ("processing_time", "desc", "B,A,C"), + ("delivery_time", "asc", "A,B,C"), + ("delivery_time", "desc", "C,B,A"), + ("processed_at", "asc", "A,B,C"), + ("processed_at", "desc", "C,B,A") + ]; + + using (Assert.EnterMultipleScope()) + { + foreach (var (sort, direction, expectedOrder) in sortOptions) + { + var result = await MessagesViewStore.GetMessages(false, new PagingInfo(), new SortInfo(sort, direction), cancellationToken: TestTimeoutCancellationToken); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(MessageIds(expectedOrder)), $"Sorted by {sort} {direction}"); + } + } + } + + [Test] + public async Task Message_list_should_be_paged_after_sorting() + { + var result = await MessagesViewStore.GetMessages(false, new PagingInfo(page: 2, pageSize: 1), new SortInfo("time_sent", "asc"), cancellationToken: TestTimeoutCancellationToken); + + using (Assert.EnterMultipleScope()) + { + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(MessageIds("B"))); + Assert.That(result.QueryStats.TotalCount, Is.EqualTo(3)); + } + } + + [Test] + public async Task Full_text_search_should_find_messages() + { + (string SearchTerm, string ExpectedMessages, string Description)[] searches = + [ + ("zebrafish", "A,C", "A word in the body"), + ("ZEBRAFISH", "A,C", "Case is ignored"), + ("zebra*", "A,C", "A trailing wildcard"), + ("bravo", "B", "A word in the body of one message"), + ("TypeB", "B", "A message type from the metadata"), + ("EndpointB", "B", "An endpoint name from the headers"), + ("no-such-term", "", "No match") + ]; + + using (Assert.EnterMultipleScope()) + { + foreach (var (searchTerm, expectedMessages, description) in searches) + { + var result = await MessagesViewStore.QueryMessages(searchTerm, new PagingInfo(), new SortInfo("time_sent", "asc"), cancellationToken: TestTimeoutCancellationToken); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(MessageIds(expectedMessages)), $"Search for '{searchTerm}': {description}"); + } + } + } + + [Test] + public async Task Full_text_search_should_combine_with_the_receiving_endpoint() + { + var endpointA = await MessagesViewStore.QueryMessagesByReceivingEndpointAndKeyword("EndpointA", "zebrafish", new PagingInfo(), new SortInfo("time_sent", "desc"), cancellationToken: TestTimeoutCancellationToken); + var endpointB = await MessagesViewStore.QueryMessagesByReceivingEndpointAndKeyword("EndpointB", "zebrafish", new PagingInfo(), new SortInfo("time_sent", "desc"), cancellationToken: TestTimeoutCancellationToken); + + using (Assert.EnterMultipleScope()) + { + Assert.That(endpointA.Results.Select(m => m.MessageId), Is.EqualTo(MessageIds("C,A"))); + Assert.That(endpointB.Results, Is.Empty); + } + } + + [Test] + public async Task Receiving_endpoint_view_should_list_the_messages_of_the_endpoint() + { + var result = await MessagesViewStore.QueryMessagesByReceivingEndpoint(false, "EndpointA", new PagingInfo(), new SortInfo("critical_time", "asc"), cancellationToken: TestTimeoutCancellationToken); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(MessageIds("C,A"))); + } + + [Test] + public async Task Conversation_view_should_list_the_messages_of_the_conversation() + { + var result = await MessagesViewStore.QueryMessagesByConversationId("conversation-1", new PagingInfo(), new SortInfo("time_sent", "desc"), TestTimeoutCancellationToken); + + Assert.That(result.Results.Select(m => m.MessageId), Is.EqualTo(MessageIds("B,A"))); + } + + [Test] + public async Task Audit_counts_should_count_the_processed_messages_of_the_endpoint() + { + var result = await MessagesViewStore.QueryAuditCounts("EndpointA", TestTimeoutCancellationToken); + + Assert.That(result.Results.Sum(c => c.Count), Is.EqualTo(2)); + } + + [Test] + public async Task Saga_view_should_contain_the_state_changes_of_the_saga_newest_first() + { + var result = await SagaHistoryStore.QuerySagaHistoryById(SagaId, TestTimeoutCancellationToken); + + Assert.That(result.Results, Is.Not.Null); + Assert.That(result.Results.Changes.Select(c => c.StateAfterChange), Is.EqualTo(new[] { "second", "first" })); + } + + static string[] MessageIds(string messages) => messages.Split(',', StringSplitOptions.RemoveEmptyEntries).Select(m => $"message-{m.ToLowerInvariant()}").ToArray(); + + async Task CreateExistingDatabaseWithCoraxIndexes() + { + var embeddedDatabase = await SharedEmbeddedServer.GetInstance(); + using var store = await embeddedDatabase.Connect(); + + await store.Maintenance.Server.SendAsync(new CreateDatabaseOperation(new DatabaseRecord(databaseName))); + + AbstractIndexCreationTask[] indexes = [new FailedAuditImportIndex(), new SagaDetailsIndex(), new MessagesViewIndexWithFullTextSearch()]; + + foreach (var index in indexes) + { + index.Configuration[IndexDeployment.StaticSearchEngineTypeKey] = nameof(SearchEngineType.Corax); + await index.ExecuteAsync(store, store.Conventions, databaseName); + } + } + + async Task IngestAudits() + { + await using var unitOfWork = await StartAuditUnitOfWork(3); + + // Each message differs in every sort field, in a different order + await RecordMessage("A", "conversation-1", "EndpointA", 0, criticalTime: 3, processingTime: 2, deliveryTime: 1, "alpha zebrafish"); + await RecordMessage("B", "conversation-1", "EndpointB", 1, criticalTime: 1, processingTime: 3, deliveryTime: 2, "bravo"); + await RecordMessage("C", "conversation-2", "EndpointA", 2, criticalTime: 2, processingTime: 1, deliveryTime: 3, "charlie zebrafish"); + + await unitOfWork.RecordSagaSnapshot(new SagaSnapshot { SagaId = SagaId, SagaType = "MySagaType", Endpoint = "EndpointA", StateAfterChange = "first", Status = SagaStateChangeStatus.New, StartTime = FirstTimeSent, FinishTime = FirstTimeSent.AddSeconds(1) }); + await unitOfWork.RecordSagaSnapshot(new SagaSnapshot { SagaId = SagaId, SagaType = "MySagaType", Endpoint = "EndpointA", StateAfterChange = "second", Status = SagaStateChangeStatus.Updated, StartTime = FirstTimeSent.AddSeconds(2), FinishTime = FirstTimeSent.AddSeconds(3) }); + + await unitOfWork.Complete(); + await configuration.CompleteDBOperation(); + + Task RecordMessage(string name, string conversationId, string endpoint, int minutesAfterFirst, int criticalTime, int processingTime, int deliveryTime, string body) + { + var timeSent = FirstTimeSent.AddMinutes(minutesAfterFirst); + var message = AuditTestData.CreateProcessedMessage( + messageId: $"message-{name.ToLowerInvariant()}", + messageType: $"Type{name}", + conversationId: conversationId, + receivingEndpoint: endpoint, + timeSent: timeSent, + criticalTime: TimeSpan.FromSeconds(criticalTime), + processingTime: TimeSpan.FromSeconds(processingTime), + deliveryTime: TimeSpan.FromSeconds(deliveryTime), + processedAt: new DateTimeOffset(timeSent)); + + return unitOfWork.RecordProcessedMessage(message, Encoding.UTF8.GetBytes(body)); + } + } + + string databaseName; +} diff --git a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs index 43524c8d80..666be0dcac 100644 --- a/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs +++ b/src/ServiceControl.Audit.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs @@ -34,19 +34,19 @@ public static async Task GetInstance(CancellationToken cancell var dbPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "Tests", "AuditData"); var logPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "Logs", "Audit"); - var logsMode = "Operations"; + var logsMinLevel = "Warn"; var serverUrl = $"http://localhost:{PortUtility.GetAssignedOrAvailablePort(33334)}"; - var databaseConfiguration = new DatabaseConfiguration("audit", 60, true, TimeSpan.FromMinutes(5), 120000, 5, 5, new ServerConfiguration(dbPath, serverUrl, logPath, logsMode), TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(30)); + var databaseConfiguration = new DatabaseConfiguration("audit", 60, true, TimeSpan.FromMinutes(5), 120000, 5, 5, new ServerConfiguration(dbPath, serverUrl, logPath, logsMinLevel), TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(30)); var serverConfig = databaseConfiguration.ServerConfiguration; // TODO: See if more refactoring can be done in configuration classes - var embeddedConfig = new EmbeddedDatabaseConfiguration(serverConfig.ServerUrl, databaseConfiguration.Name, serverConfig.DbPath, serverConfig.LogPath, serverConfig.LogsMode); + var embeddedConfig = new EmbeddedDatabaseConfiguration(serverConfig.ServerUrl, databaseConfiguration.Name, serverConfig.DbPath, serverConfig.LogPath, serverConfig.LogsMinLevel); - embeddedDatabase = EmbeddedDatabase.Start(embeddedConfig, new ApplicationLifetime(new NullLogger())); + var server = EmbeddedDatabase.Start(embeddedConfig, new ApplicationLifetime(new NullLogger())); //make sure that the database is up - using var documentStore = await embeddedDatabase.Connect(cancellationToken); + using var documentStore = await server.Connect(cancellationToken); var databaseSetup = new DatabaseSetup(databaseConfiguration); await databaseSetup.Execute(documentStore, cancellationToken); @@ -63,6 +63,7 @@ public static async Task GetInstance(CancellationToken cancell await documentStore.Maintenance.Server.SendAsync(cleanupOperation, CancellationToken.None); } + embeddedDatabase = server; return embeddedDatabase; } finally diff --git a/src/ServiceControl.Infrastructure.Tests/RavenDbLogLevelToLogsMinLevelMapperTests.cs b/src/ServiceControl.Infrastructure.Tests/RavenDbLogLevelToLogsMinLevelMapperTests.cs new file mode 100644 index 0000000000..2d58bed45f --- /dev/null +++ b/src/ServiceControl.Infrastructure.Tests/RavenDbLogLevelToLogsMinLevelMapperTests.cs @@ -0,0 +1,28 @@ +namespace ServiceControl.Infrastructure.Tests; + +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; + +[TestFixture] +public class RavenDbLogLevelToLogsMinLevelMapperTests +{ + [TestCase("Trace", "Trace")] + [TestCase("Debug", "Debug")] + [TestCase("Info", "Info")] + [TestCase("Warn", "Warn")] + [TestCase("Error", "Error")] + [TestCase("Fatal", "Fatal")] + [TestCase("Off", "Off")] + [TestCase("warn", "Warn")] + [TestCase("OFF", "Off")] + [TestCase("None", "Off")] + [TestCase("Operations", "Info")] + [TestCase("Information", "Debug")] + [TestCase("operations", "Info")] + public void Should_map_the_configured_level_to_the_server_minimum_level(string configured, string expected) => + Assert.That(RavenDbLogLevelToLogsMinLevelMapper.Map(configured, NullLogger.Instance), Is.EqualTo(expected)); + + [Test] + public void Should_fall_back_to_Warn_for_an_unknown_level() => + Assert.That(RavenDbLogLevelToLogsMinLevelMapper.Map("Verbose", NullLogger.Instance), Is.EqualTo("Warn")); +} diff --git a/src/ServiceControl.Infrastructure/RavenDbLogLevelToLogsMinLevelMapper.cs b/src/ServiceControl.Infrastructure/RavenDbLogLevelToLogsMinLevelMapper.cs new file mode 100644 index 0000000000..d1957ee0d2 --- /dev/null +++ b/src/ServiceControl.Infrastructure/RavenDbLogLevelToLogsMinLevelMapper.cs @@ -0,0 +1,38 @@ +namespace ServiceControl +{ + using Microsoft.Extensions.Logging; + + public class RavenDbLogLevelToLogsMinLevelMapper + { + public static string Map(string ravenDbLogLevel, ILogger logger) + { + switch (ravenDbLogLevel.ToLowerInvariant()) + { + case "trace": + return "Trace"; + case "debug": + return "Debug"; + case "info": + return "Info"; + case "warn": + return "Warn"; + case "error": + return "Error"; + case "fatal": + return "Fatal"; + case "off": + return "Off"; + // Legacy Logs.Mode names, translated the way RavenDB translates them + case "none": + return "Off"; + case "operations": + return "Info"; + case "information": + return "Debug"; + default: + logger.LogWarning("Unknown log level '{RavenDbLogLevel}', mapped to 'Warn'", ravenDbLogLevel); + return "Warn"; + } + } + } +} diff --git a/src/ServiceControl.Infrastructure/RavenDbLogLevelToLogsModeMapper.cs b/src/ServiceControl.Infrastructure/RavenDbLogLevelToLogsModeMapper.cs deleted file mode 100644 index d3316ba830..0000000000 --- a/src/ServiceControl.Infrastructure/RavenDbLogLevelToLogsModeMapper.cs +++ /dev/null @@ -1,30 +0,0 @@ -namespace ServiceControl -{ - using Microsoft.Extensions.Logging; - - public class RavenDbLogLevelToLogsModeMapper - { - public static string Map(string ravenDbLogLevel, ILogger logger) - { - switch (ravenDbLogLevel.ToLower()) - { - case "off": // Backwards compatibility with 4.x - case "none": - return "None"; - case "trace": // Backwards compatibility with 4.x - case "debug": // Backwards compatibility with 4.x - case "info": // Backwards compatibility with 4.x - case "information": - return "Information"; - case "error": // Backwards compatibility with 4.x - case "warn": // Backwards compatibility with 4.x - case "fatal": // Backwards compatibility with 4.x - case "operations": - return "Operations"; - default: - logger.LogWarning("Unknown log level '{RavenDbLogLevel}', mapped to 'Operations'", ravenDbLogLevel); - return "Operations"; - } - } - } -} \ No newline at end of file diff --git a/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs index 55cd75225d..5c5b078e65 100644 --- a/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Editing/EditFailedMessagesDataStore.cs @@ -2,6 +2,7 @@ namespace ServiceControl.Persistence.RavenDB.Editing { using System.Threading; using System.Threading.Tasks; + using Raven.Client.Documents.Session; using Raven.Client.Exceptions; using ServiceControl.MessageFailures; using ServiceControl.Persistence.Recoverability.Editing; @@ -18,7 +19,7 @@ public async Task GetCurrentEditingRequestId(string failedMessageId, Can public async Task TryBeginEdit(string failedMessageId, string editingMessageId, CancellationToken cancellationToken = default) { using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - session.Advanced.UseOptimisticConcurrency = true; + session.Advanced.OptimisticConcurrencyMode = OptimisticConcurrencyMode.Writes; var failedMessage = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failedMessageId), cancellationToken); if (failedMessage is null) diff --git a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs index 52bb6f3b5e..970a74f7d5 100644 --- a/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/ErrorMessagesDataStore.cs @@ -358,7 +358,7 @@ public async Task MarkAsResolved(string failedMessageId, CancellationToken var documentId = FailedMessageIdGenerator.MakeDocumentId(failedMessageId); using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - session.Advanced.UseOptimisticConcurrency = true; + session.Advanced.OptimisticConcurrencyMode = OptimisticConcurrencyMode.Writes; var failedMessage = await session.LoadAsync(documentId, cancellationToken); @@ -452,7 +452,7 @@ public async Task UnArchiveMessages(IEnumerable failedMessageI Dictionary failedMessages; using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); - session.Advanced.UseOptimisticConcurrency = true; + session.Advanced.OptimisticConcurrencyMode = OptimisticConcurrencyMode.Writes; var documentIds = failedMessageIds.Select(FailedMessageIdGenerator.MakeDocumentId); diff --git a/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs b/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs index 34af6b3f7a..2cc9207afd 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenEmbeddedPersistenceLifecycle.cs @@ -39,7 +39,7 @@ public async Task Initialize(CancellationToken cancellationToken = default) await initializeSemaphore.WaitAsync(cancellationToken); // TODO: See if more can be refactored out of RavenPersisterSettings - var embeddedConfig = new EmbeddedDatabaseConfiguration(databaseConfiguration.ServerUrl, databaseConfiguration.DatabaseName, databaseConfiguration.DatabasePath, databaseConfiguration.LogPath, databaseConfiguration.LogsMode); + var embeddedConfig = new EmbeddedDatabaseConfiguration(databaseConfiguration.ServerUrl, databaseConfiguration.DatabaseName, databaseConfiguration.DatabasePath, databaseConfiguration.LogPath, databaseConfiguration.LogsMinLevel); database = EmbeddedDatabase.Start(embeddedConfig, lifetime); diff --git a/src/ServiceControl.Persistence.RavenDB/RavenEnvironmentDataProvider.cs b/src/ServiceControl.Persistence.RavenDB/RavenEnvironmentDataProvider.cs index 41ba550345..2dde1be256 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenEnvironmentDataProvider.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenEnvironmentDataProvider.cs @@ -33,7 +33,7 @@ public IEnumerable GetData() => Deferred("Health.Error.FailedImports", FailedImports), Value("Storage.FullTextSearch", () => settings.EnableFullTextSearchOnBodies ? "Enabled" : "Disabled"), Value("Storage.BodyStorage.Type", () => "RavenAttachments"), - Value("Storage.LogLevel", () => settings.LogsMode), + Value("Storage.LogLevel", () => settings.LogsMinLevel), Value("Storage.QueryTimeoutSeconds", () => WhenConfigured(QueryTimeLimit.SettingName, () => Number((int)Math.Round(settings.QueryTimeout.TotalSeconds, MidpointRounding.AwayFromZero)))), Value("Storage.FreeSpaceThresholdPercent", () => WhenConfigured(RavenPersistenceConfiguration.DataSpaceRemainingThresholdKey, () => Number(settings.DataSpaceRemainingThreshold))), Value("Storage.MinimumFreeSpaceForIngestionPercent", () => WhenConfigured(RavenBootstrapper.MinimumStorageLeftRequiredForIngestionKey, () => Number(settings.MinimumStorageLeftRequiredForIngestion))), diff --git a/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs b/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs index 690d373b5e..15ee6b5cc4 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenExternalPersistenceLifecycle.cs @@ -5,11 +5,12 @@ namespace ServiceControl.Persistence.RavenDB using System; using System.Threading; using System.Threading.Tasks; + using Microsoft.Extensions.Logging; using Raven.Client.Documents; using Raven.Client.Documents.Conventions; using ServiceControl.RavenDB; - sealed class RavenExternalPersistenceLifecycle(RavenPersisterSettings settings) : IRavenPersistenceLifecycle, IRavenDocumentStoreProvider, IDisposable + sealed class RavenExternalPersistenceLifecycle(RavenPersisterSettings settings, ILogger logger) : IRavenPersistenceLifecycle, IRavenDocumentStoreProvider, IDisposable { public async ValueTask GetDocumentStore(CancellationToken cancellationToken = default) { @@ -42,7 +43,8 @@ public async Task Initialize(CancellationToken cancellationToken = default) Certificate = RavenClientCertificate.FindClientCertificate(settings), Conventions = new DocumentConventions { - SaveEnumsAsIntegers = true + SaveEnumsAsIntegers = true, + SaveEnumsAsIntegersForPatching = true } }; @@ -51,7 +53,7 @@ public async Task Initialize(CancellationToken cancellationToken = default) await StartupChecks.EnsureServerVersion(store, cancellationToken); var databaseSetup = new DatabaseSetup(settings, store); - await databaseSetup.Execute(cancellationToken).ConfigureAwait(false); + await DatabaseLoadRetry.Execute(databaseSetup.Execute, logger, cancellationToken); } finally { diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs index 7dfaeb715c..ee5071d7ac 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersistenceConfiguration.cs @@ -22,7 +22,7 @@ class RavenPersistenceConfiguration : PersistenceConfiguration, IPersistenceConf public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootNamespace) { var ravenDbLogLevel = SettingsReader.Read(settingsRootNamespace, RavenBootstrapper.RavenDbLogLevelKey, "Warn"); - var logsMode = RavenDbLogLevelToLogsModeMapper.Map(ravenDbLogLevel, LoggerUtil.CreateStaticLogger()); + var logsMinLevel = RavenDbLogLevelToLogsMinLevelMapper.Map(ravenDbLogLevel, LoggerUtil.CreateStaticLogger()); var settings = new RavenPersisterSettings { @@ -43,7 +43,7 @@ public PersistenceSettings CreateSettings(SettingsRootNamespace settingsRootName QueryTimeout = QueryTimeLimit.Read(settingsRootNamespace, LoggerUtil.CreateStaticLogger()), MaintenanceMode = SettingsReader.Read(settingsRootNamespace, MaintenanceModeKey, false), LogPath = SettingsReader.Read(settingsRootNamespace, RavenBootstrapper.LogsPathKey, DefaultLogLocation()), - LogsMode = logsMode, + LogsMinLevel = logsMinLevel, EnableFullTextSearchOnBodies = SettingsReader.Read(settingsRootNamespace, "EnableFullTextSearchOnBodies", true), ThroughputDatabaseName = SettingsReader.Read(ThroughputSettings.SettingsNamespace, ThroughputSettings.DatabaseNameKey, ThroughputSettings.DefaultDatabaseName) }; diff --git a/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs b/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs index 0f59fd9554..b4625773a7 100644 --- a/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs +++ b/src/ServiceControl.Persistence.RavenDB/RavenPersisterSettings.cs @@ -29,13 +29,13 @@ class RavenPersisterSettings : PersistenceSettings, IRavenClientCertificateInfo public string ClientCertificatePassword { get; set; } public bool UseEmbeddedServer => string.IsNullOrWhiteSpace(ConnectionString); public string LogPath { get; set; } - public string LogsMode { get; set; } = LogsModeDefault; + public string LogsMinLevel { get; set; } = LogsMinLevelDefault; public string DatabaseName { get; set; } = DatabaseNameDefault; public string ThroughputDatabaseName { get; set; } = ThroughputSettings.DefaultDatabaseName; public const string DatabaseNameDefault = "primary"; public const int DatabaseMaintenancePortDefault = 33334; public const int ExpirationProcessTimerInSecondsDefault = 600; - public const string LogsModeDefault = "Operations"; + public const string LogsMinLevelDefault = "Warn"; public const int ExternalIntegrationsDispatchingBatchSizeDefault = 100; } \ No newline at end of file diff --git a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs index 9171a7df6f..2294c0347a 100644 --- a/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs +++ b/src/ServiceControl.Persistence.RavenDB/Recoverability/Archiving/MessageArchiver.cs @@ -6,6 +6,7 @@ using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; + using Raven.Client.Documents.Session; using RavenDB; using ServiceControl.Infrastructure.Auth; using ServiceControl.Infrastructure.DomainEvents; @@ -45,7 +46,7 @@ public async Task ArchiveAllInGroup(string groupId, AuditUser? initiatedBy = nul using (var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken)) { - session.Advanced.UseOptimisticConcurrency = true; // Ensure 2 messages don't split the same operation into batches at once + session.Advanced.OptimisticConcurrencyMode = OptimisticConcurrencyMode.Writes; // Ensure 2 messages don't split the same operation into batches at once archiveOperation = await archiveDocumentManager.LoadArchiveOperation(session, groupId, ArchiveType.FailureGroup, cancellationToken); @@ -145,7 +146,7 @@ public async Task UnarchiveAllInGroup(string groupId, AuditUser? initiatedBy = n using (var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken)) { - session.Advanced.UseOptimisticConcurrency = true; // Ensure 2 messages don't split the same operation into batches at once + session.Advanced.OptimisticConcurrencyMode = OptimisticConcurrencyMode.Writes; // Ensure 2 messages don't split the same operation into batches at once unarchiveOperation = await unarchiveDocumentManager.LoadUnarchiveOperation(session, groupId, ArchiveType.FailureGroup, cancellationToken); diff --git a/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs b/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs index 3b67232127..bc0a8bc583 100644 --- a/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/RetryDocumentDataStore.cs @@ -29,7 +29,7 @@ public async Task AssignMessagesToBatch(string batchId, string[] messageIds, Can using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); - var batch = new SingleNodeBatchCommand(documentStore.Conventions, session.Advanced.Context, commands); + var batch = new SingleNodeBatchCommand(documentStore.Conventions, commands); await session.Advanced.RequestExecutor.ExecuteAsync(batch, session.Advanced.Context, token: cancellationToken); } diff --git a/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs b/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs index 1c64c3bc67..c2dec5ddf7 100644 --- a/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/RetryStagingStore.cs @@ -158,7 +158,7 @@ public async Task RecordStagingFailure(IReadOnlyCollection uniqueMessage using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken); var documentStore = await documentStoreProvider.GetDocumentStore(cancellationToken); - var batch = new SingleNodeBatchCommand(documentStore.Conventions, session.Advanced.Context, commands); + var batch = new SingleNodeBatchCommand(documentStore.Conventions, commands); await session.Advanced.RequestExecutor.ExecuteAsync(batch, session.Advanced.Context, token: cancellationToken); } catch (ConcurrencyException) diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/EndpointExtensions.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/EndpointExtensions.cs index ebe95db5ed..3278cdfa04 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/EndpointExtensions.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/EndpointExtensions.cs @@ -1,15 +1,20 @@ #nullable enable namespace ServiceControl.Persistence.RavenDB.Throughput; +using System.Linq; using Particular.LicensingComponent.Contracts; using ServiceControl.Persistence.RavenDB.Throughput.Models; static class EndpointExtensions { - public static string GenerateDocumentId(this EndpointIdentifier endpointId) => $"{endpointId.Name}/{endpointId.ThroughputSource}"; + public static string GenerateDocumentId(this EndpointIdentifier endpointId) => $"{EscapeControlCharacters(endpointId.Name)}/{endpointId.ThroughputSource}"; public static string GenerateDocumentId(this EndpointDocument endpoint) => endpoint.EndpointId.GenerateDocumentId(); + // Databases created by RavenDB 7.2.5 or later reject control characters in document IDs + static string? EscapeControlCharacters(string? name) => + name is not null && name.Any(char.IsControl) ? string.Concat(name.Select(c => char.IsControl(c) ? $"%{(int)c:X2}" : c.ToString())) : name; + public static Endpoint ToEndpoint(this EndpointDocument endpointDocument) => new(endpointDocument.EndpointId) { SanitizedName = endpointDocument.SanitizedName, diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/IndexDefinitionTests.Primary_index_definitions_are_unchanged.approved.txt b/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/IndexDefinitionTests.Primary_index_definitions_are_unchanged.approved.txt new file mode 100644 index 0000000000..d8a1484c04 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/ApprovalFiles/IndexDefinitionTests.Primary_index_definitions_are_unchanged.approved.txt @@ -0,0 +1,265 @@ +Index: ArchivedGroupsViewIndex + Type: MapReduce + SourceType: Documents + Map: + docs.FailedMessages.Where(doc => doc.Status == 4).Select(doc => new { + doc = doc, + failureTimes = doc.ProcessingAttempts.Select(x => x.FailureDetails.TimeOfFailure) + }).SelectMany(this0 => this0.doc.FailureGroups, (this0, failureGroup) => new { + Id = failureGroup.Id, + Title = failureGroup.Title, + Count = 1, + First = DynamicEnumerable.Min(this0.failureTimes), + Last = DynamicEnumerable.Max(this0.failureTimes), + Type = failureGroup.Type + }) + Reduce: + results.GroupBy(result => new { + Id = result.Id, + Title = result.Title, + Type = result.Type + }).Select(g => new { + Id = g.Key.Id, + Title = g.Key.Title, + Count = Enumerable.Sum(g, x => ((int) x.Count)), + First = DynamicEnumerable.Min(g, x0 => ((DateTime) x0.First)), + Last = DynamicEnumerable.Max(g, x1 => ((DateTime) x1.Last)), + Type = g.Key.Type + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: CustomChecksIndex + Type: Map + SourceType: Documents + Map: + docs.CustomChecks.Select(cc => new { + Status = cc.Status, + ReportedAt = cc.ReportedAt, + Category = cc.Category, + CustomCheckId = cc.CustomCheckId + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: EndpointIndex + Type: Map + SourceType: Documents + Map: + docs.EndpointDocuments.Select(message => new { + ThroughputSource = message.EndpointId.ThroughputSource, + Name = message.EndpointId.Name + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: FailedErrorImportIndex + Type: Map + SourceType: Documents + Map: + docs.FailedErrorImports.Select(cc => new { + Id = Id(cc) + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: FailedMessageFacetsIndex + Type: Map + SourceType: Documents + Map: + docs.FailedMessages.Where(failure => failure.Status == 1).Select(failure => new { + failure = failure, + t = ((DynamicEnumerable.LastOrDefault(failure.ProcessingAttempts)).MessageMetadata["ReceivingEndpoint"]) + }).Select(this0 => new { + Name = this0.t.Name, + Host = this0.t.Host, + MessageType = (DynamicEnumerable.LastOrDefault(this0.failure.ProcessingAttempts)).MessageMetadata["MessageType"] + }) + Field Host: {"Indexing":"Exact"} + Field MessageType: {"Indexing":"Exact"} + Field Name: {"Indexing":"Exact"} + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: FailedMessageRetries/ByBatch + Type: Map + SourceType: Documents + Map: + docs.FailedMessageRetries.Select(doc => new { + RetryBatchId = doc.RetryBatchId + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: FailedMessageViewIndex + Type: Map + SourceType: Documents + Map: + docs.FailedMessages.Select(message => new { + message = message, + processingAttemptsLast = DynamicEnumerable.LastOrDefault(message.ProcessingAttempts) + }).Select(this0 => new { + MessageId = this0.processingAttemptsLast.MessageMetadata["MessageId"], + MessageType = this0.processingAttemptsLast.MessageMetadata["MessageType"], + Status = this0.message.Status, + TimeSent = ((DateTime) this0.processingAttemptsLast.MessageMetadata["TimeSent"]), + ReceivingEndpointName = ((this0.processingAttemptsLast.MessageMetadata["ReceivingEndpoint"])).Name, + QueueAddress = this0.processingAttemptsLast.FailureDetails.AddressOfFailingEndpoint, + TimeOfFailure = this0.processingAttemptsLast.FailureDetails.TimeOfFailure, + LastModified = (this.MetadataFor(this0.message).Value < DateTime > ("@last-modified")).Ticks + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: FailedMessages/ByGroup + Type: Map + SourceType: Documents + Map: + docs.FailedMessages.Select(doc => new { + doc = doc, + processingAttemptsLast = DynamicEnumerable.LastOrDefault(doc.ProcessingAttempts) + }).SelectMany(this0 => this0.doc.FailureGroups, (this0, failureGroup) => new { + Id = Id(this0.doc), + MessageId = this0.doc.UniqueMessageId, + FailureGroupId = failureGroup.Id, + FailureGroupName = failureGroup.Title, + Status = this0.doc.Status, + MessageType = ((string) this0.processingAttemptsLast.MessageMetadata["MessageType"]), + TimeSent = ((DateTime) this0.processingAttemptsLast.MessageMetadata["TimeSent"]), + TimeOfFailure = this0.processingAttemptsLast.FailureDetails.TimeOfFailure, + LastModified = (this.MetadataFor(this0.doc).Value < DateTime > ("@last-modified")).Ticks + }) + Field __all_fields: {"Storage":"Yes"} + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: FailureGroupsViewIndex + Type: MapReduce + SourceType: Documents + Map: + docs.FailedMessages.Where(doc => doc.Status == 1).Select(doc => new { + doc = doc, + failureTimes = doc.ProcessingAttempts.Select(x => x.FailureDetails.TimeOfFailure) + }).SelectMany(this0 => this0.doc.FailureGroups, (this0, failureGroup) => new { + Id = failureGroup.Id, + Title = failureGroup.Title, + Count = 1, + First = DynamicEnumerable.Min(this0.failureTimes), + Last = DynamicEnumerable.Max(this0.failureTimes), + Type = failureGroup.Type + }) + Reduce: + results.GroupBy(result => new { + Id = result.Id, + Title = result.Title, + Type = result.Type + }).Select(g => new { + Id = g.Key.Id, + Title = g.Key.Title, + Count = Enumerable.Sum(g, x => ((int) x.Count)), + First = DynamicEnumerable.Min(g, x0 => ((DateTime) x0.First)), + Last = DynamicEnumerable.Max(g, x1 => ((DateTime) x1.Last)), + Type = g.Key.Type + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: GroupCommentIndex + Type: Map + SourceType: Documents + Map: + docs.GroupComments.Select(gc => new { + Id = Id(gc), + Comment = gc.Comment + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: KnownEndpointIndex + Type: Map + SourceType: Documents + Map: + docs.KnownEndpoints.Select(message => new { + EndpointDetails_Name = message.EndpointDetails.Name, + EndpointDetails_Host = message.EndpointDetails.Host, + HostDisplayName = message.HostDisplayName, + Monitored = message.Monitored + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: MessagesViewIndex + Type: Map + SourceType: Documents + Map: + docs.FailedMessages.Select(message => new { + message = message, + last = DynamicEnumerable.LastOrDefault(message.ProcessingAttempts) + }).Select(this0 => new { + MessageId = this0.last.MessageId, + MessageType = ((string) this0.last.MessageMetadata["MessageType"]), + IsSystemMessage = ((bool) this0.last.MessageMetadata["IsSystemMessage"]), + Status = this0.message.Status == 4 ? 5 : (this0.message.Status == 2 ? 4 : (this0.message.ProcessingAttempts.Count == 1 ? 1 : 2)), + TimeSent = ((DateTime) this0.last.MessageMetadata["TimeSent"]), + ProcessedAt = this0.last.AttemptedAt, + ReceivingEndpointName = ((this0.last.MessageMetadata["ReceivingEndpoint"])).Name, + CriticalTime = ((TimeSpan ? ) this0.last.MessageMetadata["CriticalTime"]), + ProcessingTime = ((TimeSpan ? ) this0.last.MessageMetadata["ProcessingTime"]), + DeliveryTime = ((TimeSpan ? ) this0.last.MessageMetadata["DeliveryTime"]), + Query = Enumerable.ToArray(DynamicEnumerable.Union(this0.last.MessageMetadata.ToDictionary(e1 => e1.Key, e1 => e1.Value).Select(_ => _.Value.ToString()), new string[] { + String.Join(" ", this0.last.Headers.ToDictionary(e1 => e1.Key, e1 => e1.Value).Select(x => x.Value)) + })), + ConversationId = ((string) this0.last.MessageMetadata["ConversationId"]) + }) + Field Query: {"Indexing":"Search"} + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: QueueAddressIndex + Type: MapReduce + SourceType: Documents + Map: + docs.FailedMessages.Select(message => new { + message = message, + processingAttemptsLast = DynamicEnumerable.LastOrDefault(message.ProcessingAttempts) + }).Select(this0 => new { + PhysicalAddress = this0.processingAttemptsLast.FailureDetails.AddressOfFailingEndpoint, + FailedMessageCount = 1 + }) + Reduce: + results.GroupBy(result => result.PhysicalAddress).Select(g => new { + PhysicalAddress = g.Key, + FailedMessageCount = Enumerable.Sum(g, m => ((int) m.FailedMessageCount)) + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: RetryBatches/ByStatus/ReduceInitialBatchSize + Type: MapReduce + SourceType: Documents + Map: + docs.RetryBatches.Select(doc => new { + RequestId = doc.RequestId, + RetryType = doc.RetryType, + HasStagingBatches = doc.Status == 2, + HasForwardingBatches = doc.Status == 3, + InitialBatchSize = doc.InitialBatchSize, + Originator = doc.Originator, + Classifier = doc.Classifier, + StartTime = doc.StartTime, + Last = doc.Last + }) + Reduce: + results.GroupBy(result => new { + RequestId = result.RequestId, + RetryType = result.RetryType + }).Select(g => new { + RequestId = g.Key.RequestId, + RetryType = g.Key.RetryType, + Originator = (DynamicEnumerable.FirstOrDefault(g)).Originator, + HasStagingBatches = Enumerable.Any(g, x => x.HasStagingBatches), + HasForwardingBatches = Enumerable.Any(g, x0 => x0.HasForwardingBatches), + InitialBatchSize = Enumerable.Sum(g, x1 => ((int) x1.InitialBatchSize)), + StartTime = (DynamicEnumerable.FirstOrDefault(g)).StartTime, + Last = DynamicEnumerable.Max(g, x2 => ((DateTime) x2.Last)), + Classifier = (DynamicEnumerable.FirstOrDefault(g)).Classifier + }) + Configuration Indexing.Static.SearchEngineType: Lucene + +Index: RetryBatches/ByStatusAndSession + Type: Map + SourceType: Documents + Map: + docs.RetryBatches.Select(doc => new { + RetrySessionId = doc.RetrySessionId, + Status = doc.Status + }) + Configuration Indexing.Static.SearchEngineType: Lucene + diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/DatabaseLoadRetryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/DatabaseLoadRetryTests.cs new file mode 100644 index 0000000000..5d05ce2c0a --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/DatabaseLoadRetryTests.cs @@ -0,0 +1,57 @@ +namespace ServiceControl.Persistence.Tests; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging.Abstractions; +using NUnit.Framework; +using Raven.Client.Exceptions.Database; +using ServiceControl.RavenDB; + +[TestFixture] +public class DatabaseLoadRetryTests +{ + [Test] + public async Task Should_retry_when_database_load_times_out() + { + var attempts = 0; + + await DatabaseLoadRetry.Execute(_ => + { + attempts++; + return attempts < 3 ? Task.FromException(new DatabaseLoadTimeoutException("Database is loading")) : Task.CompletedTask; + }, NullLogger.Instance, TimeSpan.FromMilliseconds(1)); + + Assert.That(attempts, Is.EqualTo(3)); + } + + [Test] + public void Should_not_retry_other_exceptions() + { + var attempts = 0; + + Assert.ThrowsAsync(() => DatabaseLoadRetry.Execute(_ => + { + attempts++; + return Task.FromException(new InvalidOperationException()); + }, NullLogger.Instance, TimeSpan.FromMilliseconds(1))); + + Assert.That(attempts, Is.EqualTo(1)); + } + + [Test] + public void Should_stop_retrying_when_cancelled() + { + using var tokenSource = new CancellationTokenSource(); + var attempts = 0; + + Assert.CatchAsync(() => DatabaseLoadRetry.Execute(_ => + { + attempts++; + tokenSource.Cancel(); + return Task.FromException(new DatabaseLoadTimeoutException("Database is loading")); + }, NullLogger.Instance, TimeSpan.FromMinutes(1), tokenSource.Token)); + + Assert.That(attempts, Is.EqualTo(1)); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/EmbeddedServerLoggingTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/EmbeddedServerLoggingTests.cs new file mode 100644 index 0000000000..9d6466eda9 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/EmbeddedServerLoggingTests.cs @@ -0,0 +1,30 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System.IO; +using System.Text.Json; +using System.Threading.Tasks; +using NUnit.Framework; + +[TestFixture] +class EmbeddedServerLoggingTests : RavenPersistenceTestBase +{ + // Asks the server instead of waiting for a log file, because at Warn a quiet server never creates one + [Test] + public async Task Server_uses_the_configured_log_path_and_minimum_level() + { + var client = DocumentStore.GetRequestExecutor().HttpClient; + + using var response = await client.GetAsync(DocumentStore.Urls[0].TrimEnd('/') + "/admin/logs/configuration"); + response.EnsureSuccessStatusCode(); + + using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); + var logs = document.RootElement.GetProperty("Logs"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(Path.GetFullPath(logs.GetProperty("Path").GetString()!).TrimEnd(Path.DirectorySeparatorChar), Is.EqualTo(Path.GetFullPath(SharedEmbeddedServer.LogPath).TrimEnd(Path.DirectorySeparatorChar))); + Assert.That(logs.GetProperty("MinLevel").GetString(), Is.EqualTo(SharedEmbeddedServer.LogsMinLevel)); + Assert.That(Directory.Exists(SharedEmbeddedServer.LogPath), Is.True); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/EnvironmentDataConfigurationTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/EnvironmentDataConfigurationTests.cs index 8f1d0d0783..00067d65d0 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/EnvironmentDataConfigurationTests.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/EnvironmentDataConfigurationTests.cs @@ -27,7 +27,7 @@ public async Task Should_report_defaults_when_nothing_is_configured() using (Assert.EnterMultipleScope()) { - Assert.That(data["Storage.LogLevel"], Is.EqualTo("Operations")); + Assert.That(data["Storage.LogLevel"], Is.EqualTo("Warn")); Assert.That(data["Storage.QueryTimeoutSeconds"], Is.EqualTo("Default")); Assert.That(data["Storage.FreeSpaceThresholdPercent"], Is.EqualTo("Default")); Assert.That(data["Storage.MinimumFreeSpaceForIngestionPercent"], Is.EqualTo("Default")); diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/ExpiresMetadataTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/ExpiresMetadataTests.cs new file mode 100644 index 0000000000..2756127ce5 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/ExpiresMetadataTests.cs @@ -0,0 +1,152 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.Expiration; + +using System; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Raven.Client; +using Raven.Client.ServerWide.Operations; +using ServiceControl.MessageFailures; +using ServiceControl.Persistence.RavenDB; +using ServiceControl.Recoverability; + +// The server deletes a document once its @expires value has passed. These tests check where ServiceControl +// puts the value on and takes it off, because the JavaScript patches that do it fail silently when they break. +[TestFixture] +class ExpiresMetadataTests : RavenPersistenceTestBase +{ + [Test] + public async Task Database_setup_configures_the_server_to_delete_expired_documents() + { + var settings = (RavenPersisterSettings)PersistenceSettings; + + var databaseRecord = await DocumentStore.Maintenance.Server.SendAsync(new GetDatabaseRecordOperation(settings.DatabaseName)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(databaseRecord.Expiration.Disabled, Is.False); + Assert.That(databaseRecord.Expiration.DeleteFrequencyInSec, Is.EqualTo(settings.ExpirationProcessTimerInSeconds)); + } + } + + [Test] + public async Task Archiving_a_group_sets_the_expiry_and_unarchiving_it_removes_it() + { + var group = new FailedMessage.FailureGroup { Id = Guid.NewGuid().ToString(), Title = "OrderPlaced", Type = "Message Type" }; + var failure = new IngestedFailure { Groups = [group] }; + var documentId = await Seed(failure, FailedMessageStatus.Unresolved, expires: false); + + await ArchiveMessages.ArchiveAllInGroup(group.Id); + var archived = await Load(documentId); + + await CompleteDatabaseOperation(); + await ArchiveMessages.UnarchiveAllInGroup(group.Id); + var unarchived = await Load(documentId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(archived.Status, Is.EqualTo(FailedMessageStatus.Archived)); + Assert.That(archived.HasExpiry, Is.True); + Assert.That(unarchived.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(unarchived.HasExpiry, Is.False); + } + } + + [Test] + public async Task Unarchiving_messages_by_id_removes_the_expiry() + { + var failure = new IngestedFailure(); + var documentId = await Seed(failure, FailedMessageStatus.Archived, expires: true); + + await FailedMessageLifecycleStore.UnArchiveMessages([failure.UniqueMessageIdString]); + + var message = await Load(documentId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(message.HasExpiry, Is.False); + } + } + + [Test] + public async Task Unarchiving_messages_by_range_unarchives_only_the_messages_in_the_range_and_removes_their_expiry() + { + var failure = new IngestedFailure(); + var documentId = await Seed(failure, FailedMessageStatus.Archived, expires: true); + var now = DateTime.UtcNow; + + var outsideRange = await FailedMessageLifecycleStore.UnArchiveMessagesByRange(now.AddHours(1), now.AddHours(2)); + var untouched = await Load(documentId); + + var insideRange = await FailedMessageLifecycleStore.UnArchiveMessagesByRange(now.AddHours(-1), now.AddHours(1)); + var unarchived = await Load(documentId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(outsideRange, Is.Empty); + Assert.That(untouched.Status, Is.EqualTo(FailedMessageStatus.Archived)); + Assert.That(untouched.HasExpiry, Is.True); + Assert.That(insideRange, Is.EqualTo(new[] { documentId })); + Assert.That(unarchived.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(unarchived.HasExpiry, Is.False); + } + } + + [Test] + public async Task Issuing_a_retry_removes_the_expiry() + { + var failure = new IngestedFailure(); + var documentId = await Seed(failure, FailedMessageStatus.Unresolved, expires: true); + + var batchId = await RetryBatchStore.CreateBatch( + RetryDocumentManager.RetrySessionId, + "request-1", + RetryType.MultipleMessages, + [failure.UniqueMessageIdString], + "a retry request", + DateTime.UtcNow); + await RetryBatchStore.AssignMessagesToBatch(batchId, [failure.UniqueMessageIdString]); + await RetryBatchStore.MoveBatchToStaging(batchId); + + await RetryStagingStore.MarkBatchAsForwarding(batchId, "staging-1", [failure.UniqueMessageIdString]); + + var message = await Load(documentId); + + using (Assert.EnterMultipleScope()) + { + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.RetryIssued)); + Assert.That(message.HasExpiry, Is.False); + } + } + + async Task Seed(IngestedFailure failure, FailedMessageStatus status, bool expires) + { + var message = failure.ToFailedMessage(status); + message.Id = PersistenceTestsContext.GenerateFailedMessageRecordId(message.UniqueMessageId); + + using (var session = await SessionProvider.OpenSession()) + { + await session.StoreAsync(message); + + if (expires) + { + session.Advanced.GetMetadataFor(message)[Constants.Documents.Metadata.Expires] = DateTime.UtcNow.AddDays(30); + } + + await session.SaveChangesAsync(); + } + + await CompleteDatabaseOperation(); + + return message.Id; + } + + async Task<(FailedMessageStatus Status, bool HasExpiry)> Load(string documentId) + { + using var session = DocumentStore.OpenAsyncSession(); + var message = await session.LoadAsync(documentId); + + return (message.Status, session.Advanced.GetMetadataFor(message).ContainsKey(Constants.Documents.Metadata.Expires)); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/IndexDefinitionTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/IndexDefinitionTests.cs new file mode 100644 index 0000000000..2ac433e684 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/IndexDefinitionTests.cs @@ -0,0 +1,115 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System; +using System.Linq; +using System.Text; +using System.Text.Json; +using System.Text.Json.Serialization; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.Approvals; +using Raven.Client.Documents.Indexes; +using ServiceControl.RavenDB; + +// RavenDB rebuilds an index side by side when the definition the client sends differs from the stored one, which +// takes days on a large database. A change to the approved file needs a plan for that rebuild. +[TestFixture] +class IndexDefinitionTests +{ + [Test] + public async Task Primary_index_definitions_are_unchanged() + { + await using var database = await ScratchDatabase.Create(); + + var indexes = PrimaryIndexes.Types.Select(PrimaryIndexes.Create).ToList(); + foreach (var index in indexes) + { + index.Conventions = database.Store.Conventions; + } + + // Deploying sets the search engine on each index. That setting is part of what the server compares. + await IndexDeployment.CreateIndexesAsync(indexes, database.Store); + + var approved = new StringBuilder(); + foreach (var index in indexes.OrderBy(index => index.IndexName, StringComparer.Ordinal)) + { + Describe(approved, index.IndexName, index.CreateIndexDefinition()); + } + + Approver.Verify(approved.ToString()); + } + + static void Describe(StringBuilder text, string name, IndexDefinition definition) + { + text.AppendLine($"Index: {name}"); + text.AppendLine($" Type: {definition.Type}"); + text.AppendLine($" SourceType: {definition.SourceType}"); + AppendIfSet(text, nameof(definition.LockMode), definition.LockMode); + AppendIfSet(text, nameof(definition.Priority), definition.Priority); + AppendIfSet(text, nameof(definition.State), definition.State); + AppendIfSet(text, nameof(definition.DeploymentMode), definition.DeploymentMode); + AppendIfSet(text, nameof(definition.ArchivedDataProcessingBehavior), definition.ArchivedDataProcessingBehavior); + + foreach (var map in definition.Maps.OrderBy(map => map, StringComparer.Ordinal)) + { + text.AppendLine(" Map:"); + AppendBlock(text, map); + } + + if (definition.Reduce is not null) + { + text.AppendLine(" Reduce:"); + AppendBlock(text, definition.Reduce); + } + + foreach (var (field, options) in definition.Fields.OrderBy(field => field.Key, StringComparer.Ordinal)) + { + text.AppendLine($" Field {field}: {JsonSerializer.Serialize(options, JsonOptions)}"); + } + + foreach (var compoundField in definition.CompoundFields ?? []) + { + text.AppendLine($" CompoundField: {string.Join(", ", compoundField)}"); + } + + foreach (var (key, value) in definition.Configuration.OrderBy(setting => setting.Key, StringComparer.Ordinal)) + { + text.AppendLine($" Configuration {key}: {value}"); + } + + foreach (var (key, value) in definition.AdditionalSources.OrderBy(source => source.Key, StringComparer.Ordinal)) + { + text.AppendLine($" AdditionalSource {key}:"); + AppendBlock(text, value); + } + + foreach (var assembly in definition.AdditionalAssemblies.Select(assembly => JsonSerializer.Serialize(assembly, JsonOptions)).OrderBy(assembly => assembly, StringComparer.Ordinal)) + { + text.AppendLine($" AdditionalAssembly: {assembly}"); + } + + AppendIfSet(text, nameof(definition.OutputReduceToCollection), definition.OutputReduceToCollection); + AppendIfSet(text, nameof(definition.ReduceOutputIndex), definition.ReduceOutputIndex); + AppendIfSet(text, nameof(definition.PatternForOutputReduceToCollectionReferences), definition.PatternForOutputReduceToCollectionReferences); + AppendIfSet(text, nameof(definition.PatternReferencesCollectionName), definition.PatternReferencesCollectionName); + text.AppendLine(); + } + + static void AppendBlock(StringBuilder text, string block) + { + foreach (var line in block.Replace("\r\n", "\n").Split('\n')) + { + text.AppendLine($" {line}"); + } + } + + static void AppendIfSet(StringBuilder text, string name, object value) + { + if (value is not null) + { + text.AppendLine($" {name}: {value}"); + } + } + + static readonly JsonSerializerOptions JsonOptions = new() { Converters = { new JsonStringEnumConverter() }, DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull }; +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/IndexDeploymentTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/IndexDeploymentTests.cs new file mode 100644 index 0000000000..c967c4bb07 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/IndexDeploymentTests.cs @@ -0,0 +1,43 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System; +using System.Collections.Generic; +using System.Threading.Tasks; +using NUnit.Framework; +using Raven.Client.Documents.Indexes; +using Raven.Client.Documents.Operations.Indexes; +using ServiceControl.RavenDB; + +// IndexDeployment deploys all indexes in one atomic call, so one index that the server cannot compile blocks +// the deployment of every other index. Each index has to deploy on its own. +[TestFixture] +class IndexDeploymentTests +{ + [TestCaseSource(nameof(IndexTypes))] + public async Task Index_deploys_to_a_clean_database(Type indexType) + { + await using var database = await ScratchDatabase.Create(); + var index = PrimaryIndexes.Create(indexType); + + Assert.That(await database.Store.Maintenance.SendAsync(new GetIndexOperation(index.IndexName)), Is.Null, "the database starts without the index"); + + await IndexDeployment.CreateIndexesAsync([index], database.Store); + + var deployed = await database.Store.Maintenance.SendAsync(new GetIndexStatisticsOperation(index.IndexName)); + + using (Assert.EnterMultipleScope()) + { + Assert.That(deployed, Is.Not.Null); + Assert.That(deployed.State, Is.EqualTo(IndexState.Normal)); + Assert.That(deployed.SearchEngineType, Is.EqualTo(SearchEngineType.Lucene)); + } + } + + public static IEnumerable IndexTypes() + { + foreach (var type in PrimaryIndexes.Types) + { + yield return new TestCaseData(type).SetArgDisplayNames(type.Name); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/MessagesViewIndexTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/MessagesViewIndexTests.cs new file mode 100644 index 0000000000..3ce26112df --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/MessagesViewIndexTests.cs @@ -0,0 +1,36 @@ +namespace ServiceControl.Persistence.Tests.RavenDB +{ + using System.Linq; + using System.Threading.Tasks; + using NUnit.Framework; + using Raven.Client.Documents; + using Raven.Client.Documents.Linq; + using Raven.Client.Documents.Operations.Indexes; + using ServiceControl.MessageFailures; + using ServiceControl.Persistence; + + [TestFixture] + class MessagesViewIndexTests : RavenPersistenceTestBase + { + [Test] + public async Task Should_map_failed_messages_without_errors_and_find_them_by_header_value() + { + var failedMessage = await SeedFailedMessage(FailedMessageBuilder.Default(m => m.ProcessingAttempts[0].Headers["HeaderA"] = "Gandalf")); + await DocumentStore.WaitForIndexingAsync(); + + var indexErrors = await DocumentStore.Maintenance.SendAsync(new GetIndexErrorsOperation([nameof(MessagesViewIndex)])); + + using var session = DocumentStore.OpenAsyncSession(); + var results = await session.Query() + .Search(x => x.Query, "Gandalf") + .OfType() + .ToListAsync(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(indexErrors.Single().Errors, Is.Empty); + Assert.That(results.Select(x => x.Id), Is.EqualTo(new[] { failedMessage.Id })); + } + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/PrimaryIndexes.cs b/src/ServiceControl.Persistence.Tests.RavenDB/PrimaryIndexes.cs new file mode 100644 index 0000000000..caf571dcfc --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/PrimaryIndexes.cs @@ -0,0 +1,18 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System; +using System.Collections.Generic; +using System.Linq; +using Raven.Client.Documents.Indexes; +using ServiceControl.Persistence.RavenDB; + +// The indexes that DatabaseSetup deploys, found the way IndexDeployment finds them. +static class PrimaryIndexes +{ + public static IReadOnlyList Types { get; } = typeof(DatabaseSetup).Assembly.GetTypes() + .Where(type => type.IsClass && !type.IsAbstract && type.IsSubclassOf(typeof(AbstractIndexCreationTask))) + .OrderBy(type => type.FullName, StringComparer.Ordinal) + .ToArray(); + + public static AbstractIndexCreationTask Create(Type type) => (AbstractIndexCreationTask)Activator.CreateInstance(type); +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/RavenServerContractTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/RavenServerContractTests.cs new file mode 100644 index 0000000000..b350a86d02 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/RavenServerContractTests.cs @@ -0,0 +1,77 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System.Text.Json; +using System.Threading.Tasks; +using Microsoft.Extensions.DependencyInjection; +using NUnit.Framework; +using ServiceControl.Persistence.RavenDB; + +// The persister reads these two endpoints into hand-written types. A field that the server renames +// deserializes to its default value instead of failing, so the checks that rely on them would pass silently. +[TestFixture] +class RavenServerContractTests : RavenPersistenceTestBase +{ + [Test] + public async Task Memory_stats_contain_the_fields_the_dirty_memory_check_reads() + { + var json = await GetJson("/admin/debug/memory/stats?includeThreads=false&includeMappings=false"); + + Assert.That(json.TryGetProperty("MemoryInformation", out var memoryInformation), Is.True); + using (Assert.EnterMultipleScope()) + { + Assert.That(memoryInformation.TryGetProperty("IsHighDirty", out var isHighDirty) && isHighDirty.ValueKind is JsonValueKind.True or JsonValueKind.False, Is.True, "IsHighDirty is a boolean"); + Assert.That(memoryInformation.TryGetProperty("DirtyMemory", out var dirtyMemory) && dirtyMemory.ValueKind == JsonValueKind.String, Is.True, "DirtyMemory is a string"); + } + } + + [Test] + public async Task Memory_information_is_read_from_the_server() + { + var retriever = ServiceProvider.GetRequiredService(); + + var (_, dirtyMemory) = await retriever.GetMemoryInformation(); + + Assert.That(dirtyMemory, Is.Not.Null.And.Not.Empty); + } + + [Test] + public async Task License_status_contains_the_fields_the_license_check_reads() + { + var json = await GetJson("/license/status"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(json.TryGetProperty("Id", out var id) && id.ValueKind == JsonValueKind.String, Is.True, "Id is a string"); + Assert.That(json.TryGetProperty("LicensedTo", out var licensedTo) && licensedTo.ValueKind == JsonValueKind.String, Is.True, "LicensedTo is a string"); + Assert.That(json.TryGetProperty("Status", out var status) && status.ValueKind == JsonValueKind.String, Is.True, "Status is a string"); + Assert.That(json.TryGetProperty("Expired", out var expired) && expired.ValueKind is JsonValueKind.True or JsonValueKind.False, Is.True, "Expired is a boolean"); + } + } + + [Test] + public async Task License_is_valid() + { + var json = await GetJson("/license/status"); + + using (Assert.EnterMultipleScope()) + { + Assert.That(json.GetProperty("Id").GetString(), Is.EqualTo("64c6a174-3f3a-4e7d-ac5d-b3eedd801460")); + Assert.That(json.GetProperty("LicensedTo").GetString(), Is.EqualTo("NserviceBus Ltd")); + Assert.That(json.GetProperty("Status").GetString(), Is.EqualTo("Commercial")); + Assert.That(json.GetProperty("Expired").GetBoolean(), Is.False); + } + + Assert.DoesNotThrowAsync(() => LicenseStatusCheck.WaitForLicenseOrThrow(DocumentStore)); + } + + async Task GetJson(string path) + { + var client = DocumentStore.GetRequestExecutor().HttpClient; + + using var response = await client.GetAsync(DocumentStore.Urls[0].TrimEnd('/') + path); + response.EnsureSuccessStatusCode(); + + using var document = JsonDocument.Parse(await response.Content.ReadAsStringAsync()); + return document.RootElement.Clone(); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RecoverabilityIngestionPatchTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RecoverabilityIngestionPatchTests.cs new file mode 100644 index 0000000000..185c8e58b4 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Recoverability/RecoverabilityIngestionPatchTests.cs @@ -0,0 +1,101 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.Recoverability; + +using System; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Raven.Client; +using ServiceControl.MessageFailures; + +// Ingestion writes failed messages through JavaScript patches. They compare the serialized AttemptedAt +// values as strings, and write the collection and the CLR type of a new document by hand. +[TestFixture] +class RecoverabilityIngestionPatchTests : RavenPersistenceTestBase +{ + [Test] + public async Task The_first_failure_creates_the_document_in_the_failed_messages_collection() + { + var failure = new IngestedFailure(); + + await Ingest(failure); + + using var session = DocumentStore.OpenAsyncSession(); + var message = await session.LoadAsync(FailedMessageIdGenerator.MakeDocumentId(failure.UniqueMessageIdString)); + var metadata = session.Advanced.GetMetadataFor(message); + + using (Assert.EnterMultipleScope()) + { + Assert.That(metadata[Constants.Documents.Metadata.Collection], Is.EqualTo("FailedMessages")); + Assert.That(metadata[Constants.Documents.Metadata.RavenClrType], Is.EqualTo(typeof(FailedMessage).AssemblyQualifiedName)); + Assert.That(message.UniqueMessageId, Is.EqualTo(failure.UniqueMessageIdString)); + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + Assert.That(message.FailureGroups.Select(group => group.Id), Is.EqualTo(failure.Groups.Select(group => group.Id))); + Assert.That(message.ProcessingAttempts, Has.Count.EqualTo(1)); + } + } + + [Test] + public async Task A_redelivered_attempt_is_stored_once() + { + var failure = new IngestedFailure(); + + await Ingest(failure); + await Ingest(failure); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + Assert.That(message.ProcessingAttempts, Has.Count.EqualTo(1)); + } + + [TestCase(TimeSpan.TicksPerMinute)] + [TestCase(1)] + public async Task An_older_attempt_arriving_late_is_stored_in_order_and_leaves_the_status_and_groups_of_the_newest(long newerByTicks) + { + var older = new IngestedFailure(); + var newerGroups = new[] { new FailedMessage.FailureGroup { Id = Guid.NewGuid().ToString(), Title = "Newer", Type = "Message Type" } }; + var newer = older.NextAttempt(older.AttemptedAt.AddTicks(newerByTicks), [.. newerGroups]); + + await Ingest(newer); + await Ingest(older); + + var message = await FailedMessageQueryStore.GetFailedMessage(older.UniqueMessageIdString); + + using (Assert.EnterMultipleScope()) + { + Assert.That(message.ProcessingAttempts.Select(attempt => attempt.AttemptedAt), Is.EqualTo(new[] { older.AttemptedAt, newer.AttemptedAt })); + Assert.That(message.FailureGroups.Select(group => group.Id), Is.EqualTo(newerGroups.Select(group => group.Id))); + Assert.That(message.Status, Is.EqualTo(FailedMessageStatus.Unresolved)); + } + } + + [TestCase(0, FailedMessageStatus.Resolved)] + [TestCase(-1, FailedMessageStatus.Unresolved)] + public async Task A_retry_resolves_the_message_only_when_it_succeeded_at_or_after_the_newest_attempt(long succeededAfterAttemptByTicks, FailedMessageStatus expected) + { + var failure = new IngestedFailure(); + await Ingest(failure); + + await using (var unitOfWork = await UnitOfWorkFactory.StartNew()) + { + await unitOfWork.Recoverability.RecordSuccessfulRetry(failure.UniqueMessageIdString, failure.AttemptedAt.AddTicks(succeededAfterAttemptByTicks)); + await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); + } + + await CompleteDatabaseOperation(); + + var message = await FailedMessageQueryStore.GetFailedMessage(failure.UniqueMessageIdString); + + Assert.That(message.Status, Is.EqualTo(expected)); + } + + async Task Ingest(IngestedFailure failure) + { + await using (var unitOfWork = await UnitOfWorkFactory.StartNew()) + { + await unitOfWork.Recoverability.RecordFailedProcessingAttempt(failure.Context, failure.ProcessingAttempt, failure.Groups); + await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken); + } + + await CompleteDatabaseOperation(); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/ScratchDatabase.cs b/src/ServiceControl.Persistence.Tests.RavenDB/ScratchDatabase.cs new file mode 100644 index 0000000000..b63605715f --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/ScratchDatabase.cs @@ -0,0 +1,54 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System; +using System.Threading; +using System.Threading.Tasks; +using Raven.Client.Documents; +using Raven.Client.Documents.Conventions; +using Raven.Client.ServerWide; +using Raven.Client.ServerWide.Operations; +using ServiceControl.RavenDB; +using TestHelper; + +// An empty database on the shared embedded server, created with the search engine settings that DatabaseSetup uses for new databases. +sealed class ScratchDatabase : IAsyncDisposable +{ + ScratchDatabase(EmbeddedDatabase server, string name, DocumentStore store) + { + this.server = server; + this.name = name; + Store = store; + } + + public IDocumentStore Store { get; } + + public static async Task Create(CancellationToken cancellationToken = default) + { + var server = await SharedEmbeddedServer.GetInstance(cancellationToken); + var name = Guid.NewGuid().ToString("n"); + + var store = new DocumentStore + { + Urls = [server.ServerUrl], + Database = name, + Conventions = new DocumentConventions { SaveEnumsAsIntegers = true, SaveEnumsAsIntegersForPatching = true } + }; + store.Initialize(); + + var databaseRecord = new DatabaseRecord(name); + databaseRecord.Settings.Add("Indexing.Auto.SearchEngineType", "Lucene"); + databaseRecord.Settings.Add("Indexing.Static.SearchEngineType", "Lucene"); + await store.Maintenance.Server.SendAsync(new CreateDatabaseOperation(databaseRecord), cancellationToken); + + return new ScratchDatabase(server, name, store); + } + + public async ValueTask DisposeAsync() + { + Store.Dispose(); + await server.DeleteDatabase(name); + } + + readonly EmbeddedDatabase server; + readonly string name; +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs b/src/ServiceControl.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs index 1bcaaf8e59..cdb2fe4799 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs +++ b/src/ServiceControl.Persistence.Tests.RavenDB/SharedEmbeddedServer.cs @@ -16,6 +16,10 @@ static class SharedEmbeddedServer { + public const string LogsMinLevel = "Warn"; + + public static string LogPath => Path.Combine(TestContext.CurrentContext.WorkDirectory, "Logs", "Primary"); + public static async Task GetInstance(CancellationToken cancellationToken = default) { if (embeddedDatabase != null) @@ -33,24 +37,24 @@ public static async Task GetInstance(CancellationToken cancell } var dbPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "Tests", "PrimaryData"); - var logPath = Path.Combine(TestContext.CurrentContext.WorkDirectory, "Logs", "Primary"); - var logsMode = "Operations"; + var logPath = LogPath; + var logsMinLevel = LogsMinLevel; var settings = new RavenPersisterSettings { DatabasePath = dbPath, LogPath = logPath, - LogsMode = logsMode, + LogsMinLevel = logsMinLevel, DatabaseMaintenancePort = PortUtility.GetAssignedOrAvailablePort(RavenPersisterSettings.DatabaseMaintenancePortDefault) }; // TODO: See if more refactoring can be done between this and the RavenPersisterSettings above - var embeddedConfig = new EmbeddedDatabaseConfiguration(settings.ServerUrl, RavenPersisterSettings.DatabaseNameDefault, dbPath, logPath, logsMode) { RunInMemory = true }; + var embeddedConfig = new EmbeddedDatabaseConfiguration(settings.ServerUrl, RavenPersisterSettings.DatabaseNameDefault, dbPath, logPath, logsMinLevel) { RunInMemory = true }; - embeddedDatabase = EmbeddedDatabase.Start(embeddedConfig, new ApplicationLifetime(new NullLogger())); + var server = EmbeddedDatabase.Start(embeddedConfig, new ApplicationLifetime(new NullLogger())); //make sure that the database is up - using var documentStore = await embeddedDatabase.Connect(cancellationToken); + using var documentStore = await server.Connect(cancellationToken); var databaseSetup = new DatabaseSetup(settings, documentStore); await databaseSetup.Execute(cancellationToken); @@ -67,6 +71,7 @@ public static async Task GetInstance(CancellationToken cancell await documentStore.Maintenance.Server.SendAsync(cleanupOperation, CancellationToken.None); } + embeddedDatabase = server; return embeddedDatabase; } finally diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/StartupChecksTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/StartupChecksTests.cs new file mode 100644 index 0000000000..179e2d919e --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/StartupChecksTests.cs @@ -0,0 +1,96 @@ +namespace ServiceControl.Persistence.Tests.RavenDB; + +using System; +using NUnit.Framework; +using ServiceControl.RavenDB; + +[TestFixture] +class StartupChecksTests +{ + static readonly Version Client727 = new(7, 2, 7); + + [TestCase("6.2.0")] + [TestCase("6.2.20")] + [TestCase("7.1.9")] + public void Should_reject_servers_with_a_lower_major_minor_version_than_the_client(string serverVersion) + { + var isValid = StartupChecks.TryValidateServerVersion(serverVersion, Client727, out var error); + + using (Assert.EnterMultipleScope()) + { + Assert.That(isValid, Is.False); + Assert.That(error, Does.Contain("expects RavenDB Server version 7.2 or higher") + .And.Contain($"the server is using {serverVersion}") + .And.Contain("servicecontrol-ravendb container")); + } + } + + [Test] + public void Should_reject_servers_with_a_lower_major_minor_version_than_a_newer_client() + { + var isValid = StartupChecks.TryValidateServerVersion("7.2.9", new Version(7, 3, 0), out var error); + + using (Assert.EnterMultipleScope()) + { + Assert.That(isValid, Is.False); + Assert.That(error, Does.Contain("expects RavenDB Server version 7.3 or higher").And.Contain("the server is using 7.2.9")); + } + } + + [TestCase("7.2.0", "7.2.0")] + [TestCase("7.2.4", "7.2.4")] + [TestCase("7.2.4-nightly-20260101-0100", "7.2.4")] + public void Should_reject_servers_with_the_known_attachments_index_corruption(string serverVersion, string reportedVersion) + { + var isValid = StartupChecks.TryValidateServerVersion(serverVersion, Client727, out var error); + + using (Assert.EnterMultipleScope()) + { + Assert.That(isValid, Is.False); + Assert.That(error, Does.Contain($"RavenDB Server {reportedVersion} can corrupt the attachments index") + .And.Contain("to 7.2.5 or later") + .And.Contain("servicecontrol-ravendb container")); + } + } + + [TestCase("7.2.5")] + [TestCase("7.2.6")] + [TestCase("7.2.7")] + [TestCase("7.2.10")] + [TestCase("7.2.7+441f983")] + [TestCase("7.2.8-nightly-20260101-0100")] + [TestCase("7.3.0")] + [TestCase("8.0.0")] + public void Should_accept_compatible_servers_without_the_known_corruption(string serverVersion) + { + var isValid = StartupChecks.TryValidateServerVersion(serverVersion, Client727, out var error); + + using (Assert.EnterMultipleScope()) + { + Assert.That(isValid, Is.True); + Assert.That(error, Is.Null); + } + } + + [TestCase("7.2.5")] + [TestCase("7.2.7")] + public void Should_not_raise_the_floor_with_the_client_patch_version(string serverVersion) + { + var isValid = StartupChecks.TryValidateServerVersion(serverVersion, new Version(7, 2, 10), out _); + + Assert.That(isValid, Is.True); + } + + [TestCase("")] + [TestCase("unknown")] + public void Should_reject_unreadable_server_versions(string serverVersion) + { + var isValid = StartupChecks.TryValidateServerVersion(serverVersion, Client727, out var error); + + using (Assert.EnterMultipleScope()) + { + Assert.That(isValid, Is.False); + Assert.That(error, Does.Contain($"could not read the server version '{serverVersion}'")); + } + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Throughput/EndpointDocumentIdTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Throughput/EndpointDocumentIdTests.cs new file mode 100644 index 0000000000..c04de4132c --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Throughput/EndpointDocumentIdTests.cs @@ -0,0 +1,39 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.Throughput; + +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; +using ServiceControl.Persistence.RavenDB.Throughput; + +[TestFixture] +class EndpointDocumentIdTests : RavenPersistenceTestBase +{ + [TestCase("Sales", "Sales/Audit")] + [TestCase("Sales.Billing@machine", "Sales.Billing@machine/Audit")] + [TestCase("Sales%01Billing", "Sales%01Billing/Audit")] + [TestCase("Sales\\u0001Billing", "Sales\\u0001Billing/Audit")] + [TestCase("Ünïcode Endpoint", "Ünïcode Endpoint/Audit")] + public void Should_not_change_the_document_id_of_a_name_without_control_characters(string name, string expectedId) => + Assert.That(new EndpointIdentifier(name, ThroughputSource.Audit).GenerateDocumentId(), Is.EqualTo(expectedId)); + + [TestCase("Sales\u0001Billing", "Sales%01Billing/Audit")] + [TestCase("Sales\tBilling", "Sales%09Billing/Audit")] + [TestCase("Sales\u007FBilling", "Sales%7FBilling/Audit")] + [TestCase("Sales\u0085Billing", "Sales%85Billing/Audit")] + [TestCase("\u0000Sales\u001F", "%00Sales%1F/Audit")] + public void Should_escape_control_characters_in_the_document_id(string name, string expectedId) => + Assert.That(new EndpointIdentifier(name, ThroughputSource.Audit).GenerateDocumentId(), Is.EqualTo(expectedId)); + + [Test] + public async Task Endpoint_name_containing_a_control_character_can_be_stored_and_read_back_in_a_new_database() + { + var endpoint = new Endpoint("Endpoint\u0001Name", ThroughputSource.Audit); + + await LicensingDataStore.SaveEndpoint(endpoint); + + var endpoints = await LicensingDataStore.GetAllEndpoints(true); + + Assert.That(endpoints.Single().Id.Name, Is.EqualTo("Endpoint\u0001Name")); + } +} diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Throughput/ThroughputTimeSeriesTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Throughput/ThroughputTimeSeriesTests.cs new file mode 100644 index 0000000000..624f260c71 --- /dev/null +++ b/src/ServiceControl.Persistence.Tests.RavenDB/Throughput/ThroughputTimeSeriesTests.cs @@ -0,0 +1,86 @@ +namespace ServiceControl.Persistence.Tests.RavenDB.Throughput; + +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using NUnit.Framework; +using Particular.LicensingComponent.Contracts; + +// The usage report is built from the incremental time series that these calls write and read +[TestFixture] +class ThroughputTimeSeriesTests : RavenPersistenceTestBase +{ + static DateOnly Today => DateOnly.FromDateTime(DateTime.UtcNow); + + [Test] + public async Task Throughput_recorded_for_the_same_day_adds_up() + { + await SaveEndpoint("Endpoint", ThroughputSource.Audit); + + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today.AddDays(-1), 30); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today, 12); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today, 30); + + var throughput = (await LicensingDataStore.GetEndpointThroughputByQueueName(["Endpoint"]))["Endpoint"].Single(); + + Assert.That(throughput, Is.EqualTo(new Dictionary + { + [Today.AddDays(-1)] = 30, + [Today] = 42 + })); + } + + [Test] + public async Task Throughput_is_read_up_to_the_last_day_asked_for() + { + await SaveEndpoint("Endpoint", ThroughputSource.Audit); + + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today.AddDays(-2), 5); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today.AddDays(-1), 6); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today, 7); + + var throughput = (await LicensingDataStore.GetEndpointThroughputByQueueName(["Endpoint"], Today))["Endpoint"].Single(); + + Assert.That(throughput.Keys, Is.EqualTo(new[] { Today.AddDays(-2), Today.AddDays(-1) })); + } + + [Test] + public async Task The_last_collected_date_of_an_endpoint_is_its_newest_recorded_day() + { + await SaveEndpoint("Endpoint", ThroughputSource.Audit); + + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today.AddDays(-5), 10); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today.AddDays(-1), 10); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today.AddDays(-3), 10); + + var endpoint = await LicensingDataStore.GetEndpoint("Endpoint", ThroughputSource.Audit); + + Assert.That(endpoint.LastCollectedDate, Is.EqualTo(Today.AddDays(-1))); + } + + [Test] + public async Task The_last_collected_date_is_read_for_every_requested_endpoint() + { + await SaveEndpoint("Endpoint", ThroughputSource.Audit); + await SaveEndpoint("Endpoint", ThroughputSource.Broker); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Audit, Today, 10); + await LicensingDataStore.RecordEndpointThroughput("Endpoint", ThroughputSource.Broker, Today, 10); + var audit = new EndpointIdentifier("Endpoint", ThroughputSource.Audit); + var broker = new EndpointIdentifier("Endpoint", ThroughputSource.Broker); + var unknown = new EndpointIdentifier("Unknown", ThroughputSource.Audit); + + var endpoints = (await LicensingDataStore.GetEndpoints([audit, broker, unknown])).ToList(); + + using (Assert.EnterMultipleScope()) + { + Assert.That(endpoints.Select(endpoint => endpoint.Id), Is.EqualTo(new[] { audit, broker, unknown })); + Assert.That(endpoints[0].Endpoint.LastCollectedDate, Is.EqualTo(Today)); + Assert.That(endpoints[1].Endpoint.LastCollectedDate, Is.EqualTo(Today)); + Assert.That(endpoints[2].Endpoint, Is.Null); + } + } + + Task SaveEndpoint(string name, ThroughputSource source) => + LicensingDataStore.SaveEndpoint(new Endpoint(name, source) { SanitizedName = name }); +} diff --git a/src/ServiceControl.RavenDB/DatabaseLoadRetry.cs b/src/ServiceControl.RavenDB/DatabaseLoadRetry.cs new file mode 100644 index 0000000000..dd114c9e0e --- /dev/null +++ b/src/ServiceControl.RavenDB/DatabaseLoadRetry.cs @@ -0,0 +1,33 @@ +namespace ServiceControl.RavenDB +{ + using System.Threading; + using Microsoft.Extensions.Logging; + using Raven.Client.Exceptions.Database; + + public static class DatabaseLoadRetry + { + public static readonly TimeSpan DefaultRetryDelay = TimeSpan.FromMilliseconds(500); + + public static Task Execute(Func action, ILogger logger, CancellationToken cancellationToken = default) => + Execute(action, logger, DefaultRetryDelay, cancellationToken); + + public static async Task Execute(Func action, ILogger logger, TimeSpan retryDelay, CancellationToken cancellationToken = default) + { + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + + try + { + await action(cancellationToken); + return; + } + catch (DatabaseLoadTimeoutException e) + { + logger.LogWarning(e, "Connecting to the RavenDB database timed out. Retrying in {RetryDelay}ms...", retryDelay.TotalMilliseconds); + await Task.Delay(retryDelay, cancellationToken); + } + } + } + } +} diff --git a/src/ServiceControl.RavenDB/EmbeddedDatabase.cs b/src/ServiceControl.RavenDB/EmbeddedDatabase.cs index efdf66b06f..14714fe4ba 100644 --- a/src/ServiceControl.RavenDB/EmbeddedDatabase.cs +++ b/src/ServiceControl.RavenDB/EmbeddedDatabase.cs @@ -15,7 +15,6 @@ namespace ServiceControl.RavenDB using Raven.Client.Documents.Conventions; using Raven.Embedded; using ServiceControl.Infrastructure; - using Sparrow.Logging; public sealed class EmbeddedDatabase : IDisposable { @@ -53,19 +52,7 @@ public static EmbeddedDatabase Start(EmbeddedDatabaseConfiguration databaseConfi var nugetPackagesPath = Path.Combine(databaseConfiguration.DbPath, "Packages", "NuGet"); - var logMode = Enum.Parse(databaseConfiguration.LogsMode); - - if (logMode == LogMode.Information) // Most verbose - { - LoggingSource.Instance.EnableConsoleLogging(); - LoggingSource.Instance.SetupLogMode( - logMode, - Path.Combine(databaseConfiguration.LogPath, "Raven.Embedded"), - retentionTime: TimeSpan.FromDays(14), - retentionSize: 1024 * 1024 * 10, - compress: false - ); - } + Directory.CreateDirectory(databaseConfiguration.LogPath); logger.LogInformation("Loading RavenDB license from {LicenseFileName}", licenseFileNameAndServerDirectory.LicenseFileName); @@ -79,7 +66,7 @@ public static EmbeddedDatabase Start(EmbeddedDatabaseConfiguration databaseConfi { CommandLineArgs = [ - $"--Logs.Mode={databaseConfiguration.LogsMode}", + $"--Logs.MinLevel={databaseConfiguration.LogsMinLevel}", // HINT: If this is not set, then Raven will pick a default location relative to the server binaries // See https://github.com/ravendb/ravendb/issues/15694 $"--Indexing.NuGetPackagesPath=\"{nugetPackagesPath}\"", @@ -173,7 +160,8 @@ public async Task Connect(CancellationToken cancellationToken = { Conventions = new DocumentConventions { - SaveEnumsAsIntegers = true + SaveEnumsAsIntegers = true, + SaveEnumsAsIntegersForPatching = true }, SkipCreatingDatabase = true }; @@ -281,7 +269,7 @@ static void RecordStartup(EmbeddedDatabaseConfiguration configuration) Database Path: {configuration.DbPath} Database Size: {ByteSize.FromBytes(dataSize).ToString("#.##", CultureInfo.InvariantCulture)} Database Folder Size: {ByteSize.FromBytes(folderSize).ToString("#.##", CultureInfo.InvariantCulture)} -RavenDB Logging Level: {configuration.LogsMode} +RavenDB Logging Level: {configuration.LogsMinLevel} -------------------------------------------------------------"; logger.LogInformation(startupMessage); diff --git a/src/ServiceControl.RavenDB/EmbeddedDatabaseConfiguration.cs b/src/ServiceControl.RavenDB/EmbeddedDatabaseConfiguration.cs index 6aec695755..6b9bbb6849 100644 --- a/src/ServiceControl.RavenDB/EmbeddedDatabaseConfiguration.cs +++ b/src/ServiceControl.RavenDB/EmbeddedDatabaseConfiguration.cs @@ -1,12 +1,12 @@ namespace ServiceControl.RavenDB { - public class EmbeddedDatabaseConfiguration(string serverUrl, string dbName, string dbPath, string logPath, string logsMode) + public class EmbeddedDatabaseConfiguration(string serverUrl, string dbName, string dbPath, string logPath, string logsMinLevel) { public string Name { get; } = dbName; public string DbPath { get; } = dbPath; public string ServerUrl { get; } = serverUrl; public string LogPath { get; } = logPath; - public string LogsMode { get; } = logsMode; + public string LogsMinLevel { get; } = logsMinLevel; public bool RunInMemory { get; set; } } diff --git a/src/ServiceControl.RavenDB/StartupChecks.cs b/src/ServiceControl.RavenDB/StartupChecks.cs index a8c44278db..8b0730f1e2 100644 --- a/src/ServiceControl.RavenDB/StartupChecks.cs +++ b/src/ServiceControl.RavenDB/StartupChecks.cs @@ -43,27 +43,55 @@ public static async Task FindIndexesUsingCorax(IDocumentStore store, s public static async Task EnsureServerVersion(IDocumentStore store, CancellationToken cancellationToken = default) { - // RavenDB compatibility policy is that the major/minor version of the server must be - // equal or higher than the client and ignores the patch version. - // - // https://docs.ravendb.net/6.2/client-api/faq/backward-compatibility/#compatibility---ravendb-42-and-higher - // - // > Starting with version 4.2, RavenDB clients are compatible with any server of their own version and higher. - // > E.g. - - // > - // > Client 4.2 is compatible with Server 4.2, Server 4.5, Server 5.2, and any other server of a higher version. - var build = await store.Maintenance.Server.SendAsync(new GetBuildNumberOperation(), cancellationToken); - var serverProductVersion = new Version(build.ProductVersion); - var clientVersion = typeof(Raven.Client.Constants).Assembly.GetCustomAttributes().First().InformationalVersion; - var parts = clientVersion.Split('.'); - var clientProductVersion = new Version($"{parts[0]}.{parts[1]}"); + if (!TryValidateServerVersion(build.FullVersion, ClientVersion, out var error)) + { + throw new Exception(error); + } + } + + public static bool TryValidateServerVersion(string serverFullVersion, Version clientVersion, out string error) + { + var clientMajorMinor = new Version(clientVersion.Major, clientVersion.Minor); + + // FullVersion can carry a suffix such as "7.2.8-nightly-20260101-0100" or "7.2.7+commit" + var numericPart = serverFullVersion.Split('-', '+')[0]; + + if (!Version.TryParse(numericPart, out var serverVersion)) + { + error = $"ServiceControl expects RavenDB Server version {clientMajorMinor} or higher, but could not read the server version '{serverFullVersion}'."; + return false; + } - if (clientProductVersion > serverProductVersion) + // RavenDB ignores the patch for compatibility: https://docs.ravendb.net/7.2/client-api/faq/backward-compatibility + if (new Version(serverVersion.Major, serverVersion.Minor) < clientMajorMinor) { - throw new Exception($"ServiceControl expects RavenDB Server version {clientProductVersion} or higher, but the server is using {serverProductVersion}."); + error = $"ServiceControl expects RavenDB Server version {clientMajorMinor} or higher, but the server is using {serverVersion}. " + + "Upgrade the external RavenDB server or the servicecontrol-ravendb container before starting ServiceControl."; + return false; } + + // Compacting a migrated database corrupts its attachments index before 7.2.5: https://github.com/ravendb/ravendb/releases/tag/7.2.5 + if (serverVersion.Major == 7 && serverVersion.Minor == 2 && serverVersion < FirstServerWithAttachmentsIndexFix) + { + error = $"RavenDB Server {serverVersion} can corrupt the attachments index of migrated databases during compaction. " + + $"Upgrade the external RavenDB server or the servicecontrol-ravendb container to {FirstServerWithAttachmentsIndexFix} or later before starting ServiceControl."; + return false; + } + + error = null; + return true; + } + + static readonly Version FirstServerWithAttachmentsIndexFix = new(7, 2, 5); + + static readonly Version ClientVersion = ParseClientVersion(); + + static Version ParseClientVersion() + { + var clientVersion = typeof(Raven.Client.Constants).Assembly.GetCustomAttributes().First().InformationalVersion; + return Version.Parse(clientVersion.Split('-', '+')[0]); } static readonly ILogger Logger = LoggerUtil.CreateStaticLogger(typeof(StartupChecks)); diff --git a/src/ServiceControlInstaller.Engine.UnitTests/Configuration/UpgradeInfoTests.cs b/src/ServiceControlInstaller.Engine.UnitTests/Configuration/UpgradeInfoTests.cs new file mode 100644 index 0000000000..72a088984e --- /dev/null +++ b/src/ServiceControlInstaller.Engine.UnitTests/Configuration/UpgradeInfoTests.cs @@ -0,0 +1,56 @@ +namespace ServiceControlInstaller.Engine.UnitTests.Configuration +{ + using NuGet.Versioning; + using NUnit.Framework; + using ServiceControlInstaller.Engine.Configuration.ServiceControl; + + [TestFixture] + public class UpgradeInfoTests + { + [TestCase("6.21.0")] + [TestCase("6.21.0-beta.1")] + [TestCase("6.21.2")] + [TestCase("6.22.0")] + [TestCase("7.0.0")] + public void Should_accept_the_latest_version_of_major_6_and_later(string current) + { + var upgradeInfo = UpgradeInfo.GetUpgradePathFor(SemanticVersion.Parse(current)); + + Assert.Multiple(() => + { + Assert.That(upgradeInfo.HasIncompatibleVersion, Is.False); + Assert.That(upgradeInfo.UpgradePath, Is.Empty); + }); + } + + [TestCase("6.0.0", "6.21.0")] + [TestCase("6.20.0", "6.21.0")] + public void Should_require_the_latest_version_of_major_6_from_an_older_6(string current, string expectedPath) + { + var upgradeInfo = UpgradeInfo.GetUpgradePathFor(SemanticVersion.Parse(current)); + + Assert.Multiple(() => + { + Assert.That(upgradeInfo.HasIncompatibleVersion, Is.True); + Assert.That(upgradeInfo.ToString(), Is.EqualTo(expectedPath)); + }); + } + + [TestCase("5.11.0", "6.21.0")] + [TestCase("5.11.11", "6.21.0")] + [TestCase("5.0.0", "5.11.0 ➡️ 6.21.0")] + [TestCase("4.33.0", "5.11.0 ➡️ 6.21.0")] + [TestCase("4.26.0", "4.33.0 ➡️ 5.11.0 ➡️ 6.21.0")] + [TestCase("1.0.0", "1.48.0 ➡️ 2.1.5 ➡️ 3.8.4 ➡️ 4.33.0 ➡️ 5.11.0 ➡️ 6.21.0")] + public void Should_keep_requiring_the_last_version_of_each_earlier_major(string current, string expectedPath) + { + var upgradeInfo = UpgradeInfo.GetUpgradePathFor(SemanticVersion.Parse(current)); + + Assert.Multiple(() => + { + Assert.That(upgradeInfo.HasIncompatibleVersion, Is.True); + Assert.That(upgradeInfo.ToString(), Is.EqualTo(expectedPath)); + }); + } + } +} diff --git a/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs b/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs index c32b2b619e..20b1f1156b 100644 --- a/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs +++ b/src/ServiceControlInstaller.Engine.UnitTests/Validation/PackagedRavenDBTests.cs @@ -21,7 +21,7 @@ namespace ServiceControlInstaller.Engine.UnitTests.Validation class PackagedRavenDBTests { [Test] - public async Task CanPersistAndIndexDocumentsWithoutNativeKafkaDependencies() + public async Task CanPersistAndIndexDocumentsWithoutExcludedFiles() { var root = Path.Combine(Path.GetTempPath(), "PackagedRavenDB", Guid.NewGuid().ToString("N")); var serverDirectory = Path.Combine(root, "Server"); @@ -43,6 +43,13 @@ public async Task CanPersistAndIndexDocumentsWithoutNativeKafkaDependencies() Assert.That(Path.Combine(serverDirectory, "libcurl.dll"), Does.Not.Exist); Assert.That(Path.Combine(serverDirectory, "librdkafka.dll"), Does.Not.Exist); Assert.That(Path.Combine(serverDirectory, "librdkafkacpp.dll"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "onnxruntime.dll"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "onnxruntime.lib"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "onnxruntime_providers_shared.dll"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "onnxruntime_providers_shared.lib"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "onnxruntime-genai.dll"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "onnxruntime-genai.lib"), Does.Not.Exist); + Assert.That(Path.Combine(serverDirectory, "LocalEmbeddings"), Does.Not.Exist); Assert.That(Path.Combine(serverDirectory, "Raven.Server.exe"), Does.Exist); } diff --git a/src/ServiceControlInstaller.Engine/Configuration/ServiceControl/UpgradeInfo.cs b/src/ServiceControlInstaller.Engine/Configuration/ServiceControl/UpgradeInfo.cs index eb4a738a8d..d2280bdca0 100644 --- a/src/ServiceControlInstaller.Engine/Configuration/ServiceControl/UpgradeInfo.cs +++ b/src/ServiceControlInstaller.Engine/Configuration/ServiceControl/UpgradeInfo.cs @@ -11,7 +11,8 @@ public class UpgradeInfo new(2, 1, 5), new(3, 8, 4), new(4, 33, 0), // Added IngestErrorMessages setting to enable message replay during 4to5 upgrade guide scenario - new(5, 11, 0) + new(5, 11, 0), + new(6, 21, 0) ]; public SemanticVersion[] UpgradePath { get; private init; } diff --git a/src/ServiceControlInstaller.Packaging/ServiceControlInstaller.Packaging.csproj b/src/ServiceControlInstaller.Packaging/ServiceControlInstaller.Packaging.csproj index 7a4136f323..b0b7ce6df3 100644 --- a/src/ServiceControlInstaller.Packaging/ServiceControlInstaller.Packaging.csproj +++ b/src/ServiceControlInstaller.Packaging/ServiceControlInstaller.Packaging.csproj @@ -46,7 +46,11 @@ + Exclude="$(ArtifactsPath)$(RavenDBServer)\libcurl.dll;$(ArtifactsPath)$(RavenDBServer)\librdkafka.dll;$(ArtifactsPath)$(RavenDBServer)\librdkafkacpp.dll; + $(ArtifactsPath)$(RavenDBServer)\onnxruntime.dll;$(ArtifactsPath)$(RavenDBServer)\onnxruntime.lib; + $(ArtifactsPath)$(RavenDBServer)\onnxruntime_providers_shared.dll;$(ArtifactsPath)$(RavenDBServer)\onnxruntime_providers_shared.lib; + $(ArtifactsPath)$(RavenDBServer)\onnxruntime-genai.dll;$(ArtifactsPath)$(RavenDBServer)\onnxruntime-genai.lib; + $(ArtifactsPath)$(RavenDBServer)\LocalEmbeddings\**" />