diff --git a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs index 46639e6305..cd5e67c868 100644 --- a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs @@ -179,6 +179,15 @@ public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, Cancella return Task.CompletedTask; } + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) + { + foreach (var id in endpointIds) + { + endpoints.Remove(id); + } + return Task.CompletedTask; + } + class EndpointCollection : KeyedCollection { protected override EndpointIdentifier GetKeyForItem(Endpoint item) => item.Id; diff --git a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs index 3a73a50891..d2121fcde5 100644 --- a/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs +++ b/src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs @@ -15,6 +15,8 @@ public interface ILicensingDataStore Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken); + Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken); + Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken); Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken) => diff --git a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs index 88a22d2cfd..4fa629dfdc 100644 --- a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs @@ -203,6 +203,48 @@ await Task.Run(async () => } } + [Test] + public async Task Should_only_create_new_endpoint_when_audit_counts_exist() + { + // Arrange + using var tokenSource = new CancellationTokenSource(TimeSpan.FromSeconds(3)); + var token = tokenSource.Token; + var fakeTimeProvider = new FakeTimeProvider(); + + var date = DateOnly.FromDateTime(DateTime.UtcNow.AddDays(-1)); + var auditQuery = new AuditQuery_WithTwoEndpointsAndSelectiveCounts( + endpointWithoutCounts: "EndpointNoData", + endpointWithCounts: "EndpointWithData", + throughputDate: date, + throughputCount: 5); + + using var auditThroughputCollectorHostedService = new AuditThroughputCollectorHostedService( + NullLogger.Instance, configuration.ThroughputSettings, DataStore, + auditQuery, fakeTimeProvider) + { DelayStart = TimeSpan.Zero }; + + // Act + await auditThroughputCollectorHostedService.StartAsync(token); + await Task.Run(async () => + { + do + { + await Task.Delay(TimeSpan.FromMilliseconds(50)); + } while (!token.IsCancellationRequested); + }); + await auditThroughputCollectorHostedService.StopAsync(token); + + var endpointWithoutCounts = await DataStore.GetEndpoint("EndpointNoData", ThroughputSource.Audit, default); + var endpointWithCounts = await DataStore.GetEndpoint("EndpointWithData", ThroughputSource.Audit, default); + + // Assert + using (Assert.EnterMultipleScope()) + { + Assert.That(endpointWithoutCounts, Is.Null, "Endpoint with empty auditCounts should not be created"); + Assert.That(endpointWithCounts, Is.Not.Null, "Endpoint with auditCounts should be created"); + } + } + class AuditQuery_NoAuditRemotes : IAuditQuery { public SemanticVersion MinAuditCountsVersion => new(4, 29, 0); @@ -335,4 +377,50 @@ public string SanitizeEndpointName(string endpointName) public string SanitizedEndpointNameCleanser(string endpointName) => endpointName; } + + class AuditQuery_WithTwoEndpointsAndSelectiveCounts : IAuditQuery + { + public AuditQuery_WithTwoEndpointsAndSelectiveCounts( + string endpointWithoutCounts, + string endpointWithCounts, + DateOnly throughputDate, + long throughputCount) + { + this.endpointWithoutCounts = endpointWithoutCounts; + this.endpointWithCounts = endpointWithCounts; + this.throughputDate = throughputDate; + this.throughputCount = throughputCount; + } + + public SemanticVersion MinAuditCountsVersion => new(4, 29, 0); + public Func ValidRemoteInstances => _ => true; + + public Task> GetKnownEndpoints(CancellationToken cancellationToken) => + Task.FromResult>( + [ + new ServiceControlEndpoint { Name = endpointWithoutCounts, HeartbeatsEnabled = true }, + new ServiceControlEndpoint { Name = endpointWithCounts, HeartbeatsEnabled = true } + ]); + + public Task> GetAuditCountForEndpoint(string endpointUrlName, CancellationToken cancellationToken) + { + if (endpointUrlName == endpointWithCounts) + { + return Task.FromResult>([new AuditCount { UtcDate = throughputDate, Count = throughputCount }]); + } + + return Task.FromResult>([]); + } + + public Task> GetAuditRemotes(CancellationToken cancellationToken) => + Task.FromResult>([]); + + public Task TestAuditConnection(CancellationToken cancellationToken) => + Task.FromResult(new ConnectionSettingsTestResult { ConnectionSuccessful = true, ConnectionErrorMessages = [] }); + + readonly string endpointWithoutCounts; + readonly string endpointWithCounts; + readonly DateOnly throughputDate; + readonly long throughputCount; + } } \ No newline at end of file diff --git a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs index 6603abcad0..c991f87e6c 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs @@ -194,7 +194,7 @@ await DataStore.CreateBuilder() public async Task Should_return_correct_throughput_in_report_when_endpoint_has_no_throughput() { // Arrange - await DataStore.CreateBuilder().AddEndpoint().Build(); + await DataStore.CreateBuilder().AddEndpoint().WithThroughput(ThroughputSource.Broker, data: [0]).Build(); // Act var report = await ThroughputCollector.GenerateThroughputReport("", null, default); @@ -211,6 +211,20 @@ public async Task Should_return_correct_throughput_in_report_when_endpoint_has_n } } + [Test] + public async Task Should_not_return_endpoint_in_report_when_endpoint_has_no_throughput() + { + // Arrange + await DataStore.CreateBuilder().AddEndpoint().Build(); + + // Act + var report = await ThroughputCollector.GenerateThroughputReport("", null, default); + + // Assert + Assert.That(report, Is.Not.Null); + Assert.That(report.ReportData.Queues.Count, Is.Zero, "Invalid number of endpoints in throughput report"); + } + [Test] public async Task Should_return_correct_throughput_in_report_when_data_from_multiple_sources_and_name_is_different() { diff --git a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs index b3007fe41f..53aa86ae08 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs @@ -177,10 +177,11 @@ await DataStore.CreateBuilder() } [Test] - public async Task Should_return_correct_max_daily_throughput_in_summary_when_endpoint_has_no_throughput() + public async Task Should_return_correct_max_daily_throughput_in_summary_when_endpoint_has_zero_throughput() { // Arrange - await DataStore.CreateBuilder().AddEndpoint().Build(); + await DataStore.CreateBuilder().AddEndpoint().WithThroughput(new ThroughputData([ + new EndpointDailyThroughput(new DateOnly(2025, 1, 10), 0)])).Build(); // Act var summary = await ThroughputCollector.GetThroughputSummary(default); @@ -191,6 +192,20 @@ public async Task Should_return_correct_max_daily_throughput_in_summary_when_end Assert.That(summary[0].MaxDailyThroughput, Is.EqualTo(0), $"Incorrect MaxDailyThroughput recorded for {summary[0].Name}"); } + [Test] + public async Task Should_not_return_endpoint_in_summary_when_endpoint_has_no_throughput() + { + // Arrange + await DataStore.CreateBuilder().AddEndpoint().Build(); + + // Act + var summary = await ThroughputCollector.GetThroughputSummary(default); + + // Assert + Assert.That(summary, Is.Not.Null); + Assert.That(summary, Is.Empty, "Invalid number of endpoints in throughput summary"); + } + [Test] public async Task Should_return_correct_max_daily_throughput_in_summary_when_data_from_multiple_sources_and_name_is_different() { diff --git a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs index 91c90dbb95..dbb8b46e30 100644 --- a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs +++ b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs @@ -74,11 +74,15 @@ async Task GatherThroughput(CancellationToken cancellationToken) var auditCounts = (await auditQuery.GetAuditCountForEndpoint(knownEndpointsLookup[endpointId].UrlName, cancellationToken)).ToList(); - if (endpoint == null) + if (endpoint == null && auditCounts.Count > 0) { endpoint = ConvertToEndpoint(knownEndpointsLookup[endpointId]); await dataStore.SaveEndpoint(endpoint, cancellationToken); } + else if (endpoint is null) + { + continue; + } var missingAuditThroughput = auditCounts .Where(auditCount => auditCount.UtcDate > endpoint.LastCollectedDate && diff --git a/src/Particular.LicensingComponent/ThroughputCollector.cs b/src/Particular.LicensingComponent/ThroughputCollector.cs index 32f70aaf40..04edae5b59 100644 --- a/src/Particular.LicensingComponent/ThroughputCollector.cs +++ b/src/Particular.LicensingComponent/ThroughputCollector.cs @@ -224,7 +224,14 @@ async IAsyncEnumerable GetDistinctEndpointData([EnumeratorCancella var userIndicator = UserIndicator(endpointGroupPerQueue) ?? null; - yield return new EndpointData(endpointName, throughputData, userIndicator, EndpointScope(endpointGroupPerQueue), EndpointIndicators(endpointGroupPerQueue), IsKnownEndpoint(endpointGroupPerQueue)); + if (throughputData.Any(x => x.Any())) + { + yield return new EndpointData(endpointName, throughputData, userIndicator, EndpointScope(endpointGroupPerQueue), EndpointIndicators(endpointGroupPerQueue), IsKnownEndpoint(endpointGroupPerQueue)); + } + else + { + await dataStore.RemoveEndpoints([.. endpointGroupPerQueue.Select(endpoint => endpoint.Id)], cancellationToken); + } } } diff --git a/src/Particular.LicensingComponent/ThroughputDataExtensions.cs b/src/Particular.LicensingComponent/ThroughputDataExtensions.cs index ce19589692..18a11ed2f6 100644 --- a/src/Particular.LicensingComponent/ThroughputDataExtensions.cs +++ b/src/Particular.LicensingComponent/ThroughputDataExtensions.cs @@ -51,5 +51,5 @@ public static long AverageMonthlyThroughput(this List throughput } public static bool HasDataFromSource(this IDictionary> throughputPerQueue, ThroughputSource source) => - throughputPerQueue.Any(queueName => queueName.Value.Any(data => data.ThroughputSource == source && data.Count > 0)); + throughputPerQueue.Any(queueThroughput => queueThroughput.Value.Any(data => data.ThroughputSource == source && data.Count > 0)); } \ No newline at end of file diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs index 6e348f08c4..fa9a3bc75f 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndex.cs @@ -21,13 +21,14 @@ from message in messages TimeSent = (DateTime)message.MessageMetadata["TimeSent"], ProcessedAt = message.ProcessedAt, ReceivingEndpointName = ((EndpointDetails)message.MessageMetadata["ReceivingEndpoint"]).Name, + SendingEndpointName = ((EndpointDetails)message.MessageMetadata["SendingEndpoint"]).Name, 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(), + string.Join(" ", message.Headers.Select(x => x.Value)) + }).ToArray(), ConversationId = (string)message.MessageMetadata["ConversationId"] }; @@ -48,6 +49,7 @@ public class SortAndFilterOptions public MessageStatus Status { get; set; } public DateTime ProcessedAt { get; set; } public string ReceivingEndpointName { get; set; } + public string SendingEndpointName { get; set; } public TimeSpan? CriticalTime { get; set; } public TimeSpan? ProcessingTime { get; set; } public TimeSpan? DeliveryTime { get; set; } diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs index 9eb433d6e0..1dca30993f 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/Indexes/MessagesViewIndexWithFullTextSearch.cs @@ -21,14 +21,15 @@ from message in messages TimeSent = (DateTime)message.MessageMetadata["TimeSent"], ProcessedAt = message.ProcessedAt, ReceivingEndpointName = ((EndpointDetails)message.MessageMetadata["ReceivingEndpoint"]).Name, + SendingEndpointName = ((EndpointDetails)message.MessageMetadata["SendingEndpoint"]).Name, 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(), + string.Join(" ", message.Headers.Select(x => x.Value)), + LoadAttachment(message, "body").GetContentAsString() + }).ToArray(), ConversationId = (string)message.MessageMetadata["ConversationId"] }; diff --git a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs index 97f90af67c..88d9b56ba6 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -138,6 +138,9 @@ public async Task>> QueryAuditCounts(string endpoi .OrderBy(m => m.ProcessedAt) .FirstOrDefaultAsync(token: cancellationToken); + var hasSent = await session.Query(indexName) + .AnyAsync(m => m.SendingEndpointName == endpointName, token: cancellationToken); + if (oldestMsg != null) { var endDate = DateTime.UtcNow.Date.AddDays(1); @@ -166,6 +169,14 @@ public async Task>> QueryAuditCounts(string endpoi } } } + else if (hasSent) + { + results.Add(new AuditCount + { + UtcDate = DateTime.UtcNow.Date, + Count = 0 + }); + } return new QueryResult>(results, QueryStatsInfo.Zero); } diff --git a/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs b/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs index e067bd33b9..3344510c40 100644 --- a/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs +++ b/src/ServiceControl.Audit.Persistence.Tests/AuditCountingTests.cs @@ -67,7 +67,36 @@ string ScrubDates(string input) }, ScrubDates); } - ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bool systemMessage) + [Test] + public async Task Should_return_zero_throughput_entry_when_SendOnly() + { + // Arrange + var today = DateTime.UtcNow.Date; + const string sendOnlyEndpoint = "SendOnlyEndpoint"; + + var messages = new[] + { + // Endpoint sent a message, but did not receive any + MakeMessage("SomeOtherEndpoint", sendOnlyEndpoint, today, false) + }; + + await IngestProcessedMessagesAudits(messages); + + // Act + var result = (await DataStore.QueryAuditCounts(sendOnlyEndpoint, TestContext.CurrentContext.CancellationToken)).Results; + + // Assert + Assert.That(result, Is.Not.Empty, "Expected non-empty result for endpoint that only sent messages"); + Assert.That(result, Has.Count.EqualTo(1), "Expected single audit count for send-only endpoint"); + using (Assert.EnterMultipleScope()) + { + Assert.That(result[0].UtcDate, Is.EqualTo(today), "Expected today's date placeholder"); + Assert.That(result[0].Count, Is.Zero, "Expected zero throughput count for send-only endpoint"); + } + } + + static ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bool systemMessage) => MakeMessage(processingEndpoint, null, processedAt, systemMessage); + static ProcessedMessage MakeMessage(string processingEndpoint, string sendingEndpoint, DateTime processedAt, bool systemMessage) { var messageId = Guid.NewGuid().ToString(); var messageType = "MyMessageType"; @@ -85,8 +114,12 @@ ProcessedMessage MakeMessage(string processingEndpoint, DateTime processedAt, bo { "MessageType", messageType }, { "IsRetried", false }, { "ConversationId", messageId }, - { "ReceivingEndpoint", new EndpointDetails { Name = processingEndpoint } } + { "ReceivingEndpoint", new EndpointDetails { Name = processingEndpoint } }, }; + if (!string.IsNullOrEmpty(sendingEndpoint)) + { + metadata.Add("SendingEndpoint", new EndpointDetails { Name = sendingEndpoint }); + } var headers = new Dictionary { diff --git a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs index acf8fe9102..f57591a034 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -14,6 +14,7 @@ class LicensingDataStore : ILicensingDataStore public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, IList throughput, CancellationToken cancellationToken) => throw new NotImplementedException(); + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken) => throw new NotImplementedException(); public Task SaveEndpoint(Particular.LicensingComponent.Contracts.Endpoint endpoint, CancellationToken cancellationToken) => throw new NotImplementedException(); diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index 26b666cdcb..dc0b269a85 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -25,6 +25,8 @@ class LicensingDataStore( const string ReportMasksDocumentId = "ReportMasks"; const string LicencedEndpointDetailsDocumentId = "LicensedEndpointDetails"; + const int ThroughputPeriodMonths = 14; + static readonly AuditServiceMetadata DefaultAuditServiceMetadata = new([], []); static readonly BrokerMetadata DefaultBrokerMetadata = new(null, []); static readonly ReportConfigurationDocument DefaultReportConfiguration = new(); @@ -121,6 +123,20 @@ public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellation await session.SaveChangesAsync(cancellationToken); } + public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken) + { + var documentIds = endpointIds.Select(id => id.GenerateDocumentId()); + + var store = await storeProvider.GetDocumentStore(cancellationToken); + using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); + + foreach (var documentId in documentIds) + { + session.Delete(documentId); + } + await session.SaveChangesAsync(cancellationToken); + } + public async Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken) { var results = queueNames.ToDictionary(queueName => queueName, _ => new List() as IEnumerable); @@ -128,7 +144,7 @@ public async Task>> GetEndpointT var store = await storeProvider.GetDocumentStore(cancellationToken); using IAsyncDocumentSession session = store.OpenAsyncSession(databaseConfiguration.Name); - var from = DateTime.UtcNow.AddMonths(-14); + var from = DateTime.UtcNow.AddMonths(-ThroughputPeriodMonths); var query = session.Query() .Where(document => document.SanitizedName.In(queueNames)) .Include(builder => builder.IncludeTimeSeries(ThroughputTimeSeriesName, from)); diff --git a/src/ServiceControl/Licensing/LicenseController.cs b/src/ServiceControl/Licensing/LicenseController.cs index 51efd95cc3..cd9580773c 100644 --- a/src/ServiceControl/Licensing/LicenseController.cs +++ b/src/ServiceControl/Licensing/LicenseController.cs @@ -1,10 +1,8 @@ #nullable enable namespace ServiceControl.Licensing { - using System; using System.IO; using System.IO.Compression; - using System.Text; using System.Text.Json; using System.Threading; using System.Threading.Tasks;