Skip to content

Spike: shrink audit full-text search index #5834

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ramonsmits wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

Spike: shrink audit full-text search index #5834

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
603c4aa
✨ Exclude temporal, numeric and boolean values from audit full-text s…
ramonsmits Aug 28, 2026
7763356
✨ Match message and conversation ids exactly instead of via full-text…
ramonsmits Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions ...ntrol.Audit.AcceptanceTests/Auditing/When_processed_message_searched_by_conversationid.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
namespace ServiceControl.Audit.AcceptanceTests.Auditing
{
using System.Threading.Tasks;
using AcceptanceTesting;
using AcceptanceTesting.EndpointTemplates;
using Audit.Auditing.MessagesView;
using NServiceBus;
using NServiceBus.AcceptanceTesting;
using NServiceBus.AcceptanceTesting.Customization;
using NServiceBus.Settings;
using NUnit.Framework;

class When_processed_message_searched_by_conversationid : AcceptanceTest
{
[Test]
public async Task Should_be_found() =>
await Define<MyContext>()
.WithEndpoint<Sender>(b => b.When((bus, c) => bus.Send(new MyMessage())))
.WithEndpoint<Receiver>()
.Done(async c => c.ConversationId != null && await this.TryGetMany<MessagesView>("/api/messages/search/" + c.ConversationId))
.Run();

public class Sender : EndpointConfigurationBuilder
{
public Sender() =>
EndpointSetup<DefaultServerWithoutAudit>(c =>
{
var routing = c.ConfigureRouting();
routing.RouteToEndpoint(typeof(MyMessage), typeof(Receiver));
});
}

public class Receiver : EndpointConfigurationBuilder
{
public Receiver() => EndpointSetup<DefaultServerWithAudit>();

[Handler]
public class MyMessageHandler(MyContext testContext, IReadOnlySettings settings)
: IHandleMessages<MyMessage>
{
public Task Handle(MyMessage message, IMessageHandlerContext context)
{
testContext.EndpointNameOfReceivingEndpoint = settings.EndpointName();
testContext.ConversationId = context.MessageHeaders[Headers.ConversationId];
return Task.CompletedTask;
}
}
}

public class MyMessage : ICommand;

public class MyContext : ScenarioContext
{
public string ConversationId { get; set; }

public string EndpointNameOfReceivingEndpoint { get; set; }
}
}
}
32 changes: 23 additions & 9 deletions src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,20 +25,34 @@ from message in messages
CriticalTime = (TimeSpan?)message.MessageMetadata["CriticalTime"],
ProcessingTime = (TimeSpan?)message.MessageMetadata["ProcessingTime"],
DeliveryTime = (TimeSpan?)message.MessageMetadata["DeliveryTime"],
Query = message.MessageMetadata.Select(_ => _.Value.ToString()).Union(new[]
{
string.Join(" ", message.Headers.Select(x => x.Value))
}).ToArray(),
// Dates, durations, sizes and booleans only add tokens nobody searches for, so they are excluded.
// Identifiers are excluded too: they are matched exactly on the MessageId/ConversationId fields instead.
Query = message.MessageMetadata
.Where(m => m.Key != "MessageId" && m.Key != "ConversationId" && m.Key != "RelatedToId"
&& m.Key != "TimeSent" && m.Key != "CriticalTime" && m.Key != "ProcessingTime"
&& m.Key != "DeliveryTime" && m.Key != "ContentLength" && m.Key != "BodyUrl"
&& m.Key != "IsSystemMessage" && m.Key != "IsRetried" && m.Key != "BodyNotStored"
&& m.Key != "OriginatesFromSaga")
.Select(m => m.Value.ToString())
.Concat(message.Headers
.Where(h => h.Key != "NServiceBus.MessageId" && h.Key != "NServiceBus.ConversationId"
&& h.Key != "NServiceBus.CorrelationId" && h.Key != "NServiceBus.RelatedTo"
&& h.Key != "NServiceBus.TimeSent" && h.Key != "NServiceBus.ProcessingStarted" && h.Key != "NServiceBus.ProcessingEnded"
&& h.Key != "NServiceBus.DeliverAt" && h.Key != "NServiceBus.Timeout.Expire" && h.Key != "NServiceBus.Retries.Timestamp"
&& h.Key != "NServiceBus.ExceptionInfo.TimeOfFailure" && h.Key != "NServiceBus.TimeOfFailure" && h.Key != "NServiceBus.NonDurableMessage"
&& h.Key != "NServiceBus.TimeToBeReceived")
.Select(h => h.Value))
.Where(v => v != null && v.Length > 0)
.Distinct()
.ToArray(),
ConversationId = (string)message.MessageMetadata["ConversationId"]
};

Index(x => x.Query, FieldIndexing.Search);

// Not using typeof() to prevent dependency on Lucene.
// Unfortunately while "StandardAnalyzer" would probably be better and more future-proof here,
// we can't change this string without causing any existing audit database to completely rebuild this index.
// If this index *must* be changed for some other reason, the analyzer name should be changed at the same time.
Analyze(x => x.Query, "Lucene.Net.Analysis.Standard.StandardAnalyzer, Lucene.Net, Version=3.0.3.0, Culture=neutral, PublicKeyToken=85089178b9ac3181");
// Any change to this index definition (map or analyzer) causes existing audit databases to rebuild the index on startup.
// The analyzer name deliberately does not use typeof() to prevent a dependency on Lucene.
Analyze(x => x.Query, "StandardAnalyzer");
}

public class SortAndFilterOptions
Expand Down
34 changes: 24 additions & 10 deletions src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,21 +25,35 @@ from message in messages
CriticalTime = (TimeSpan?)message.MessageMetadata["CriticalTime"],
ProcessingTime = (TimeSpan?)message.MessageMetadata["ProcessingTime"],
DeliveryTime = (TimeSpan?)message.MessageMetadata["DeliveryTime"],
Query = message.MessageMetadata.Select(_ => _.Value.ToString()).Union(new[]
{
string.Join(" ", message.Headers.Select(x => x.Value)),
LoadAttachment(message, "body").GetContentAsString()
}).ToArray(),
// Dates, durations, sizes and booleans only add tokens nobody searches for, so they are excluded.
// Identifiers are excluded too: they are matched exactly on the MessageId/ConversationId fields instead.
Query = message.MessageMetadata
.Where(m => m.Key != "MessageId" && m.Key != "ConversationId" && m.Key != "RelatedToId"
&& m.Key != "TimeSent" && m.Key != "CriticalTime" && m.Key != "ProcessingTime"
&& m.Key != "DeliveryTime" && m.Key != "ContentLength" && m.Key != "BodyUrl"
&& m.Key != "IsSystemMessage" && m.Key != "IsRetried" && m.Key != "BodyNotStored"
&& m.Key != "OriginatesFromSaga")
.Select(m => m.Value.ToString())
.Concat(message.Headers
.Where(h => h.Key != "NServiceBus.MessageId" && h.Key != "NServiceBus.ConversationId"
&& h.Key != "NServiceBus.CorrelationId" && h.Key != "NServiceBus.RelatedTo"
&& h.Key != "NServiceBus.TimeSent" && h.Key != "NServiceBus.ProcessingStarted" && h.Key != "NServiceBus.ProcessingEnded"
&& h.Key != "NServiceBus.DeliverAt" && h.Key != "NServiceBus.Timeout.Expire" && h.Key != "NServiceBus.Retries.Timestamp"
&& h.Key != "NServiceBus.ExceptionInfo.TimeOfFailure" && h.Key != "NServiceBus.TimeOfFailure" && h.Key != "NServiceBus.NonDurableMessage"
&& h.Key != "NServiceBus.TimeToBeReceived")
.Select(h => h.Value))
.Where(v => v != null && v.Length > 0)
.Distinct()
.Concat(new[] { LoadAttachment(message, "body").GetContentAsString() })
.ToArray(),
ConversationId = (string)message.MessageMetadata["ConversationId"]
};

Index(x => x.Query, FieldIndexing.Search);

// Not using typeof() to prevent dependency on Lucene.
// Unfortunately while "StandardAnalyzer" would probably be better and more future-proof here,
// we can't change this string without causing any existing audit database to completely rebuild this index.
// If this index *must* be changed for some other reason, the analyzer name should be changed at the same time.
Analyze(x => x.Query, "Lucene.Net.Analysis.Standard.StandardAnalyzer, Lucene.Net, Version=3.0.3.0, Culture=neutral, PublicKeyToken=85089178b9ac3181");
// Any change to this index definition (map or analyzer) causes existing audit databases to rebuild the index on startup.
// The analyzer name deliberately does not use typeof() to prevent a dependency on Lucene.
Analyze(x => x.Query, "StandardAnalyzer");
}
}
}
20 changes: 16 additions & 4 deletions src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,8 @@
using Extensions;
using Indexes;
using Raven.Client.Documents;
using Raven.Client.Documents.Linq;
using Raven.Client.Documents.Session;
using ServiceControl.Audit.Auditing;
using ServiceControl.Audit.Infrastructure;
using ServiceControl.SagaAudit;
Expand Down Expand Up @@ -46,9 +48,8 @@ public async Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSyst
public async Task<QueryResult<IList<MessagesView>>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
{
using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
var results = await session.Query<MessagesViewIndex.SortAndFilterOptions>(GetIndexName(isFullTextSearchEnabled))
var results = await SearchMessages(session, searchParam)
.Statistics(out var stats)
.Search(x => x.Query, searchParam)
.FilterBySentTimeRange(timeSentRange)
.Sort(sortInfo)
.Paging(pagingInfo)
Expand All @@ -61,9 +62,8 @@ public async Task<QueryResult<IList<MessagesView>>> QueryMessages(string searchP
public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpointAndKeyword(string endpoint, string keyword, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
{
using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
var results = await session.Query<MessagesViewIndex.SortAndFilterOptions>(GetIndexName(isFullTextSearchEnabled))
var results = await SearchMessages(session, keyword)
.Statistics(out var stats)
.Search(x => x.Query, keyword)
.Where(m => m.ReceivingEndpointName == endpoint)
.FilterBySentTimeRange(timeSentRange)
.Sort(sortInfo)
Expand All @@ -74,6 +74,18 @@ public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndp
return new QueryResult<IList<MessagesView>>(results, stats.ToQueryStatsInfo());
}

// Identifiers are not part of the full-text Query field; they are matched exactly on their own fields.
// Built as a DocumentQuery because the LINQ provider does not parenthesize an OR group, so subsequent Where
// clauses would bind to the last OR term only.
IRavenQueryable<MessagesViewIndex.SortAndFilterOptions> SearchMessages(IAsyncDocumentSession session, string searchParam) =>
session.Advanced.AsyncDocumentQuery<MessagesViewIndex.SortAndFilterOptions>(GetIndexName(isFullTextSearchEnabled))
.OpenSubclause()
.Search(x => x.Query, searchParam)
.OrElse().WhereEquals(x => x.MessageId, searchParam)
.OrElse().WhereEquals(x => x.ConversationId, searchParam)
.CloseSubclause()
.ToQueryable();

public async Task<QueryResult<IList<MessagesView>>> QueryMessagesByReceivingEndpoint(bool includeSystemMessages, string endpointName, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange, CancellationToken cancellationToken = default)
{
using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
Expand Down
Loading