diff --git a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs index aa13b9852f..55f14310ac 100644 --- a/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs +++ b/src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs @@ -1,6 +1,7 @@ namespace Particular.LicensingComponent.UnitTests; using System; +using System.Collections.Concurrent; using System.Collections.Generic; using System.Collections.ObjectModel; using System.Linq; @@ -13,6 +14,7 @@ namespace Particular.LicensingComponent.UnitTests; using Infrastructure; using MonitoringThroughput; using NUnit.Framework; +using Persistence; using ServiceControl.Transports.BrokerThroughput; [TestFixture] @@ -159,6 +161,110 @@ public async Task Should_return_error_monitoring_connection_and_diagnostics_if_n Approver.Verify(connectionSettingsResult.Diagnostics); } + [Test] + public async Task Should_record_throughput_for_every_endpoint_in_the_message() + { + // Arrange + var dataStore = new YieldingLicensingDataStore(); + var monitoringService = new MonitoringService(dataStore); + + string[] endpointNames = [.. Enumerable.Range(1, 20).Select(i => $"Endpoint{i}")]; + var message = new RecordEndpointThroughputData + { + StartDateTime = DateTime.UtcNow.AddMinutes(-5), + EndDateTime = DateTime.UtcNow, + EndpointThroughputData = [.. endpointNames.Select(name => new EndpointThroughputData { Name = name, Throughput = 15 })] + }; + + // Act + byte[] messageBytes = JsonSerializer.SerializeToUtf8Bytes(message); + await monitoringService.RecordMonitoringThroughput(messageBytes); + + // Assert + string[] endpointsWithoutThroughput = [.. endpointNames.Where(name => !dataStore.RecordedThroughput.ContainsKey(name))]; + string[] unsavedEndpoints = [.. endpointNames.Where(name => !dataStore.SavedEndpoints.ContainsKey(name))]; + + using (Assert.EnterMultipleScope()) + { + Assert.That(unsavedEndpoints, Is.Empty, + $"{unsavedEndpoints.Length} of {endpointNames.Length} endpoints were not saved when RecordMonitoringThroughput returned: {string.Join(", ", unsavedEndpoints)}"); + Assert.That(endpointsWithoutThroughput, Is.Empty, + $"{endpointsWithoutThroughput.Length} of {endpointNames.Length} endpoints had no throughput recorded when RecordMonitoringThroughput returned: {string.Join(", ", endpointsWithoutThroughput)}"); + } + + Assert.That(dataStore.RecordedThroughput.Values, Is.All.EqualTo(15L), "Expected a throughput of 15 for every endpoint"); + } + + // Records only after suspending, so any write the service starts without awaiting is still + // outstanding when RecordMonitoringThroughput returns. + class YieldingLicensingDataStore : ILicensingDataStore + { + public ConcurrentDictionary SavedEndpoints { get; } = new(); + + public ConcurrentDictionary RecordedThroughput { get; } = new(); + + public async Task GetEndpoint(EndpointIdentifier id, CancellationToken cancellationToken = default) + { + await Task.Yield(); + return SavedEndpoints.TryGetValue(id.Name, out Endpoint endpoint) ? endpoint : null; + } + + public async Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken = default) + { + await Task.Yield(); + SavedEndpoints[endpoint.Id.Name] = endpoint; + } + + public async Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, + IList throughput, CancellationToken cancellationToken = default) + { + await Task.Yield(); + RecordedThroughput[endpointName] = throughput.Sum(t => t.MessageCount); + } + + public Task> GetAllEndpoints(bool includePlatformEndpoints, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task> GetEndpoints(IList endpointIds, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task>> GetEndpointThroughputByQueueName(IList queueNames, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task UpdateUserIndicatorOnEndpoints(List userIndicatorUpdates, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task IsThereThroughputForLastXDays(int days, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task IsThereThroughputForLastXDaysForSource(int days, ThroughputSource throughputSource, bool includeToday, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task GetBrokerMetadata(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveBrokerMetadata(BrokerMetadata brokerMetadata, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task GetAuditServiceMetadata(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveAuditServiceMetadata(AuditServiceMetadata auditServiceMetadata, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task> GetReportMasks(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveReportMasks(List reportMasks, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task GetLicensedEndpointDetails(CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + + public Task SaveLicensedEndpointDetails(LicensedEndpointDetails result, CancellationToken cancellationToken = default) => + throw new NotSupportedException(); + } + class BrokerThroughputQuery_WithSanitization : IBrokerThroughputQuery { public Dictionary Data => throw new NotImplementedException(); diff --git a/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs b/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs index b0dd964a06..d27ec786f3 100644 --- a/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs +++ b/src/Particular.LicensingComponent/MonitoringThroughput/MonitoringService.cs @@ -18,11 +18,11 @@ public async Task RecordMonitoringThroughput(byte[] throughputMessage, Cancellat message = await JsonSerializer.DeserializeAsync(stream, cancellationToken: cancellationToken); } - if (message != null && message.EndpointThroughputData != null) + if (message is not null && message.EndpointThroughputData is { } endpointThroughputData) { - Debug.WriteLine($"Throughput data from {message.StartDateTime:yyyy-MM-dd HH:mm} to {message.EndDateTime:yyyy-MM-dd HH:mm} for {message.EndpointThroughputData?.Length} endpoint(s)"); + Debug.WriteLine($"Throughput data from {message.StartDateTime:yyyy-MM-dd HH:mm} to {message.EndDateTime:yyyy-MM-dd HH:mm} for {endpointThroughputData.Length} endpoint(s)"); - message.EndpointThroughputData?.ToList().ForEach(async e => + foreach (var e in endpointThroughputData) { var endpoint = await dataStore.GetEndpoint(e.Name, ThroughputSource.Monitoring, cancellationToken); if (endpoint == null) @@ -41,7 +41,7 @@ public async Task RecordMonitoringThroughput(byte[] throughputMessage, Cancellat await dataStore.RecordEndpointThroughput(e.Name, ThroughputSource.Monitoring, [endpointThroughput], cancellationToken); } - }); + } } }