diff --git a/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs b/src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs index 0b7c941202..5d63fe73ec 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 = default) + { + 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 135115d51d..567ae4089c 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 = default); + Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default); + Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default); Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken = default) => diff --git a/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/AuditThroughputCollectorHostedService_Tests.cs index 3c2fdb5474..e548393052 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, CancellationToken.None); + var endpointWithCounts = await DataStore.GetEndpoint("EndpointWithData", ThroughputSource.Audit, CancellationToken.None); + + // 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 = default) => + Task.FromResult>( + [ + new ServiceControlEndpoint { Name = endpointWithoutCounts, HeartbeatsEnabled = true }, + new ServiceControlEndpoint { Name = endpointWithCounts, HeartbeatsEnabled = true } + ]); + + public Task> GetAuditCountForEndpoint(string endpointUrlName, CancellationToken cancellationToken = default) + { + if (endpointUrlName == endpointWithCounts) + { + return Task.FromResult>([new AuditCount { UtcDate = throughputDate, Count = throughputCount }]); + } + + return Task.FromResult>([]); + } + + public Task> GetAuditRemotes(CancellationToken cancellationToken = default) => + Task.FromResult>([]); + + public Task TestAuditConnection(CancellationToken cancellationToken = default) => + 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/MonitoringService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs index 55f14310ac..438cb729dd 100644 --- a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs @@ -263,6 +263,7 @@ public Task GetLicensedEndpointDetails(CancellationToke public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) => throw new NotSupportedException(); + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) => throw new NotImplementedException(); } class BrokerThroughputQuery_WithSanitization : IBrokerThroughputQuery 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 ac4503a4ca..5ae45f22a0 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs @@ -195,7 +195,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); @@ -212,6 +212,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, CancellationToken.None); + + // 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 e012779408..ddb3725bec 100644 --- a/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_ThroughputSummary_Tests.cs @@ -177,13 +177,14 @@ 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(); + var summary = await ThroughputCollector.GetThroughputSummary(CancellationToken.None); // Assert Assert.That(summary, Is.Not.Null); @@ -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(); + + // 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 8b50f6cece..5b345182dd 100644 --- a/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs +++ b/src/Particular.LicensingComponent/AuditThroughput/AuditThroughputCollectorHostedService.cs @@ -78,8 +78,13 @@ async Task GatherThroughput(CancellationToken cancellationToken) var auditCounts = (await auditQuery.GetAuditCountForEndpoint(knownEndpointsLookup[endpointId].UrlName, cancellationToken)).ToList(); - if (endpoint == null) + if (endpoint is null) { + if (auditCounts.Count <= 0) + { + continue; + } + endpoint = ConvertToEndpoint(knownEndpointsLookup[endpointId]); await dataStore.SaveEndpoint(endpoint, cancellationToken); } diff --git a/src/Particular.LicensingComponent/ThroughputCollector.cs b/src/Particular.LicensingComponent/ThroughputCollector.cs index 03f11f2a06..aa6607d85f 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.InMemory/InMemoryAuditDataStore.cs b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs index 81281db096..1ea4212771 100644 --- a/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs @@ -173,6 +173,7 @@ Task GetMessageBodyFromMetadata(string messageId, CancellationT public Task>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken = default) { + var hasSent = messageViews.Any(m => m.SendingEndpoint?.Name == endpointName); var results = messageViews .Where(m => m.ReceivingEndpoint.Name == endpointName && !m.IsSystemMessage) .GroupBy(m => m.ProcessedAt.ToUniversalTime().Date) @@ -184,7 +185,13 @@ public Task>> QueryAuditCounts(string endpointName .OrderBy(r => r.UtcDate) .ToList(); - return Task.FromResult(new QueryResult>(results, QueryStatsInfo.Zero)); + return results.Count == 0 && hasSent + ? Task.FromResult(new QueryResult>([new AuditCount + { + UtcDate = messageViews.First().ProcessedAt.ToUniversalTime().Date, + Count = 0 + }], QueryStatsInfo.Zero)) + : Task.FromResult(new QueryResult>(results, QueryStatsInfo.Zero)); } public Task SaveProcessedMessage(ProcessedMessage processedMessage, CancellationToken cancellationToken = default) 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 12a88109c6..6c39ad103c 100644 --- a/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs +++ b/src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs @@ -166,6 +166,20 @@ public async Task>> QueryAuditCounts(string endpoi } } } + else + { + var hasSent = await session.Query(indexName) + .AnyAsync(m => m.SendingEndpointName == endpointName, token: cancellationToken); + + 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 d8d02540fc..6486fa0fdc 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, DateTimeOffset processedAt, bool systemMessage) + [Test] + public async Task Should_return_zero_throughput_entry_when_SendOnly() + { + // Arrange + var today = new DateTimeOffset(DateTime.UtcNow.Date, TimeSpan.Zero); + 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.Date), "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, DateTimeOffset processedAt, bool systemMessage) => MakeMessage(processingEndpoint, null, processedAt, systemMessage); + static ProcessedMessage MakeMessage(string processingEndpoint, string sendingEndpoint, DateTimeOffset processedAt, bool systemMessage) { var messageId = Guid.NewGuid().ToString(); var messageType = "MyMessageType"; @@ -85,8 +114,12 @@ ProcessedMessage MakeMessage(string processingEndpoint, DateTimeOffset processed { "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 36538858c3..694f828209 100644 --- a/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs @@ -246,6 +246,8 @@ static async Task TryRecordEndpointThroughput(ServiceControlDbContext cont }); } + public Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) => throw new NotImplementedException(); + public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) => ExecuteWithDbContext(async (context, token) => { diff --git a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs index 4c76b83e64..ef4f23402d 100644 --- a/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs +++ b/src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs @@ -121,6 +121,20 @@ public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellation await session.SaveChangesAsync(cancellationToken); } + public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default) + { + 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 = default) { var results = queueNames.ToDictionary(queueName => queueName, _ => new List() as IEnumerable);