Don't include throughput values in throughput report after report end date

src/Particular.LicensingComponent.Persistence.InMemory/InMemoryLicensingDataStore.cs

@@ -68,7 +68,7 @@ public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken
         return Task.CompletedTask;
     }
 
-    public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken = default)
+    public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, DateOnly? throughputMaxDate = null, CancellationToken cancellationToken = default)
     {
         var result = endpoints
             .Where(endpoint => queueNames.Contains(endpoint.SanitizedName))
@@ -77,7 +77,11 @@ public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThrough
                 throughputDictionary => throughputDictionary.Key,
                 (endpoint, throughputDictionary) => new { endpoint.SanitizedName, throughputDictionary.Value })
             .GroupBy(anon => anon.SanitizedName)
-            .ToDictionary(group => group.Key, group => group.Select(entry => entry.Value));
+            .ToDictionary(group => group.Key, group => group.Select(
+                entry => new ThroughputData(entry.Value.Where(edt => !throughputMaxDate.HasValue || edt.Key < throughputMaxDate.Value).Select(edt => new EndpointDailyThroughput(edt.Key, edt.Value)))
+                {
+                    ThroughputSource = entry.Value.ThroughputSource
+                }));
 
         return Task.FromResult((IDictionary<string, IEnumerable<ThroughputData>>)result);
     }

src/Particular.LicensingComponent.Persistence/ILicensingDataStore.cs

@@ -17,7 +17,7 @@ public interface ILicensingDataStore
 
     Task RemoveEndpoints(EndpointIdentifier[] endpointIds, CancellationToken cancellationToken = default);
 
-    Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken = default);
+    Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, DateOnly? throughputMaxDate = null, CancellationToken cancellationToken = default);
 
     Task RecordEndpointThroughput(string endpointName, ThroughputSource throughputSource, DateOnly date, long messageCount, CancellationToken cancellationToken = default) =>
         RecordEndpointThroughput(endpointName, throughputSource, [new EndpointDailyThroughput(date, messageCount)], cancellationToken);

src/Particular.LicensingComponent.UnitTests/ApprovalFiles/ThroughputCollector_Report_Throughput_Tests.Should_generate_correct_report.approved.txt

@@ -7,8 +7,8 @@
     "ToolVersion": "5.0.1",
     "ScopeType": "testingScope",
     "StartTime": "2024-04-24T00:00:00+00:00",
-    "EndTime": "2024-04-25T00:00:00+00:00",
-    "ReportDuration": "1.00:00:00",
+    "EndTime": "2024-04-26T00:00:00+00:00",
+    "ReportDuration": "2.00:00:00",
     "Queues": [
       {
         "QueueName": "REDACTED1",

src/Particular.LicensingComponent.UnitTests/MonitoringService_Tests.cs

@@ -228,7 +228,7 @@ public Task<IEnumerable<Endpoint>> GetAllEndpoints(bool includePlatformEndpoints
         public Task<IEnumerable<(EndpointIdentifier Id, Endpoint Endpoint)>> GetEndpoints(IList<EndpointIdentifier> endpointIds, CancellationToken cancellationToken = default) =>
             throw new NotSupportedException();
 
-        public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken = default) =>
+        public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, DateOnly? throughputMaxDate = null, CancellationToken cancellationToken = default) =>
             throw new NotSupportedException();
 
         public Task UpdateUserIndicatorOnEndpoints(List<UpdateUserIndicator> userIndicatorUpdates, CancellationToken cancellationToken = default) =>

src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Dates_Tests.cs

@@ -2,7 +2,6 @@
 
 using System;
 using System.Linq;
-using System.Threading;
 using System.Threading.Tasks;
 using NUnit.Framework;
 using Particular.LicensingComponent.Contracts;
@@ -22,7 +21,8 @@ public override Task Setup()
     public async Task Should_return_correct_dates_for_report_when_multiple_sources_with_different_dates()
     {
         // Arrange
-        var maxDate = DateOnly.FromDateTime(DateTime.UtcNow).AddDays(-1);
+        var today = DateTime.UtcNow.Date;
+        var maxDate = DateOnly.FromDateTime(today).AddDays(-1);
         var minDate = maxDate.AddDays(-4);
 
         await DataStore.CreateBuilder()
@@ -46,7 +46,7 @@ await DataStore.CreateBuilder()
 
         // Assert
         var minDateInReport = new DateTimeOffset(minDate.ToDateTime(TimeOnly.MinValue, DateTimeKind.Utc));
-        var reportEndDate = new DateTimeOffset(maxDate.ToDateTime(TimeOnly.MinValue, DateTimeKind.Utc));
+        var reportEndDate = new DateTimeOffset(today);
 
         Assert.That(report, Is.Not.Null);
         using (Assert.EnterMultipleScope())

src/Particular.LicensingComponent.UnitTests/ThroughputCollector/ThroughputCollector_Report_Throughput_Tests.cs

@@ -286,6 +286,44 @@ await DataStore.CreateBuilder()
         }
     }
 
+    [TestCase(ThroughputSource.Audit)]
+    [TestCase(ThroughputSource.Broker)]
+    [TestCase(ThroughputSource.Monitoring)]
+    public async Task Should_not_include_throughput_after_report_end_date(ThroughputSource source)
+    {
+        // Arrange
+        var reportEndDate = new DateTime(2024, 4, 25, 0, 0, 0, DateTimeKind.Utc);
+
+        await DataStore.CreateBuilder()
+            .AddEndpoint("Endpoint1", sources: [source])
+            .WithThroughput(
+                startDate: DateOnly.FromDateTime(reportEndDate).AddDays(-2),
+                data: [50, 55, 100])
+            .Build();
+
+        // Act
+        var report = await ThroughputCollector.GenerateThroughputReport("", reportEndDate);
+
+        // Assert
+        var queue = report.ReportData.Queues.Single();
+        var dailyThroughput = source switch
+        {
+            ThroughputSource.Audit => queue.DailyThroughputFromAudit,
+            ThroughputSource.Broker => queue.DailyThroughputFromBroker,
+            ThroughputSource.Monitoring => queue.DailyThroughputFromMonitoring,
+            _ => throw new ArgumentOutOfRangeException(nameof(source))
+        };
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(dailyThroughput, Has.Length.EqualTo(2));
+            Assert.That(dailyThroughput, Has.All.Matches<DailyThroughput>(
+                throughput => throughput.DateUTC <= DateOnly.FromDateTime(reportEndDate)));
+            Assert.That(queue.Throughput, Is.EqualTo(55));
+            Assert.That(report.ReportData.TotalThroughput, Is.EqualTo(55));
+        }
+    }
+
     [Test]
     public async Task Should_generate_correct_report()
     {
@@ -324,7 +362,7 @@ await DataStore.CreateBuilder()
         await DataStore.SaveAuditServiceMetadata(new AuditServiceMetadata(expectedAuditVersionSummary, expectedAuditTransportSummary));
 
         // Act
-        var report = await ThroughputCollector.GenerateThroughputReport("2.3.1", new DateTime(2024, 4, 25));
+        var report = await ThroughputCollector.GenerateThroughputReport("2.3.1", new DateTime(2024, 4, 26));
         var reportString = System.Text.Json.JsonSerializer.Serialize(report, SerializationOptions.IndentedWithNoEscaping);
 
         // Assert

src/Particular.LicensingComponent/ThroughputCollector.cs

@@ -65,7 +65,7 @@ public async Task<List<EndpointThroughputSummary>> GetThroughputSummary(Cancella
     {
         var endpointSummaries = new List<EndpointThroughputSummary>();
 
-        await foreach (var endpointData in GetDistinctEndpointData(cancellationToken))
+        await foreach (var endpointData in GetDistinctEndpointData(null, cancellationToken))
         {
             var endpointSummary = new EndpointThroughputSummary
             {
@@ -120,10 +120,15 @@ public async Task<SignedReport> GenerateThroughputReport(string spVersion, DateT
         var reportMasks = await dataStore.GetReportMasks(cancellationToken);
         var masker = new Masker([.. reportMasks]);
 
+        if (reportEndDate is null || reportEndDate == DateTime.MinValue)
+        {
+            reportEndDate = DateTime.UtcNow.Date;
+        }
+
         var queueThroughputs = new List<QueueThroughput>();
         List<string> ignoredQueueNames = [];
 
-        await foreach (var endpointData in GetDistinctEndpointData(cancellationToken))
+        await foreach (var endpointData in GetDistinctEndpointData(DateOnly.FromDateTime(reportEndDate.Value), cancellationToken))
         {
             var notAnNsbEndpoint = endpointData.UserIndicator?.Equals(Contracts.UserIndicator.NotNServiceBusEndpoint.ToString(), StringComparison.OrdinalIgnoreCase) ?? false;
 
@@ -148,10 +153,6 @@ public async Task<SignedReport> GenerateThroughputReport(string spVersion, DateT
 
         var auditServiceMetadata = await dataStore.GetAuditServiceMetadata(cancellationToken);
         var brokerMetaData = await dataStore.GetBrokerMetadata(cancellationToken);
-        if (reportEndDate is null || reportEndDate == DateTime.MinValue)
-        {
-            reportEndDate = DateTime.UtcNow.Date.AddDays(-1);
-        }
         var report = new Report.Report
         {
             EndTime = new DateTimeOffset((DateTime)reportEndDate, TimeSpan.Zero),
@@ -197,11 +198,13 @@ public async Task<SignedReport> GenerateThroughputReport(string spVersion, DateT
         return throughputReport;
     }
 
-    async IAsyncEnumerable<EndpointData> GetDistinctEndpointData([EnumeratorCancellation] CancellationToken cancellationToken)
+    async IAsyncEnumerable<EndpointData> GetDistinctEndpointData(DateOnly? throughputMaxDate, [EnumeratorCancellation] CancellationToken cancellationToken)
     {
         var endpoints = (await dataStore.GetAllEndpoints(false, cancellationToken)).ToArray();
         var queueNames = endpoints.Select(endpoint => endpoint.SanitizedName).Distinct().ToList();
-        var endpointThroughputPerQueue = await dataStore.GetEndpointThroughputByQueueName(queueNames, cancellationToken);
+        //Some brokers will have throughput data for "today" when a throughput report is being run only expecting data up until the end of "yesterday".
+        //  Provide throughputMaxDate so that throughput from the non-complete "today" can be filtered out from the resulting ThroughputData constructs
+        var endpointThroughputPerQueue = await dataStore.GetEndpointThroughputByQueueName(queueNames, throughputMaxDate, cancellationToken);
 
         systemHasAuditEnabled = endpointThroughputPerQueue.HasDataFromSource(ThroughputSource.Audit);
         systemHasMonitoringEnabled = endpointThroughputPerQueue.HasDataFromSource(ThroughputSource.Monitoring);

src/ServiceControl.Persistence.EFCore/Implementation/LicensingDataStore.cs

@@ -112,7 +112,7 @@ public Task SaveEndpoint(Endpoint endpoint, CancellationToken cancellationToken
                 token);
         }, cancellationToken);
 
-    public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken = default) =>
+    public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, DateOnly? throughputMaxDate = null, CancellationToken cancellationToken = default) =>
         ExecuteWithDbContext(async (context, token) =>
         {
             var results = queueNames.ToDictionary(queueName => queueName, _ => Enumerable.Empty<ThroughputData>());
@@ -148,6 +148,7 @@ public Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThrough
             foreach (var endpointRows in rows.GroupBy(row => (row.NormalizedSanitizedName, row.ThroughputSource)))
             {
                 var throughputData = new ThroughputData(endpointRows
+                    .Where(row => !throughputMaxDate.HasValue || row.DateUtc < throughputMaxDate.Value)
                     .OrderBy(row => row.DateUtc)
                     .Select(row => new EndpointDailyThroughput(row.DateUtc, row.MessageCount)))
                 {

src/ServiceControl.Persistence.RavenDB/Throughput/LicensingDataStore.cs

@@ -135,7 +135,7 @@ public async Task RemoveEndpoints(EndpointIdentifier[] endpointIds, Cancellation
         await session.SaveChangesAsync(cancellationToken);
     }
 
-    public async Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, CancellationToken cancellationToken = default)
+    public async Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointThroughputByQueueName(IList<string> queueNames, DateOnly? throughputMaxDate = null, CancellationToken cancellationToken = default)
     {
         var results = queueNames.ToDictionary(queueName => queueName, _ => new List<ThroughputData>() as IEnumerable<ThroughputData>);
 
@@ -155,10 +155,13 @@ public async Task<IDictionary<string, IEnumerable<ThroughputData>>> GetEndpointT
                 .IncrementalTimeSeriesFor(document.GenerateDocumentId(), ThroughputTimeSeriesName)
                 .GetAsync(from, token: cancellationToken);
 
+            var maxDateTime = throughputMaxDate?.ToDateTime(TimeOnly.MinValue);
             if (timeSeries is not null && results.TryGetValue(document.SanitizedName, out var throughputDatas) &&
                 throughputDatas is List<ThroughputData> throughputDataList)
             {
-                var endpointDailyThroughputs = timeSeries.Select(entry => new EndpointDailyThroughput(DateOnly.FromDateTime(entry.Timestamp), (long)entry.Value));
+                var endpointDailyThroughputs = timeSeries
+                    .Where(entry => !throughputMaxDate.HasValue || entry.Timestamp < maxDateTime)
+                    .Select(entry => new EndpointDailyThroughput(DateOnly.FromDateTime(entry.Timestamp), (long)entry.Value));
                 var throughputData = new ThroughputData(endpointDailyThroughputs)
                 {
                     ThroughputSource = document.EndpointId.ThroughputSource

src/ServiceControl.Persistence.Tests/Throughput/EndpointsTests.cs

@@ -2,7 +2,6 @@
 
 using System;
 using System.Linq;
-using System.Threading;
 using System.Threading.Tasks;
 using NUnit.Framework;
 using Particular.LicensingComponent.Contracts;