Merge pull request #5619 from Particular/john/last_seen

Retire known-endpoints scatter-gather and audit KnownEndpoints storage

Author: johnsimons

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditDataStore.cs

@@ -9,15 +9,11 @@
     using ServiceControl.Audit.Auditing.BodyStorage;
     using ServiceControl.Audit.Auditing.MessagesView;
     using ServiceControl.Audit.Infrastructure;
-    using ServiceControl.Audit.Monitoring;
-    using ServiceControl.Audit.Persistence.Infrastructure;
-    using ServiceControl.Audit.Persistence.Monitoring;
     using ServiceControl.SagaAudit;
 
     class InMemoryAuditDataStore : IAuditDataStore
     {
         IBodyStorage bodyStorage;
-        public List<KnownEndpoint> knownEndpoints;
         public List<FailedAuditImport> failedAuditImports;
 
         public InMemoryAuditDataStore(IBodyStorage bodyStorage)
@@ -26,7 +22,6 @@ public InMemoryAuditDataStore(IBodyStorage bodyStorage)
             sagaHistories = [];
             messageViews = [];
             processedMessages = [];
-            knownEndpoints = [];
             failedAuditImports = [];
         }
 
@@ -176,25 +171,6 @@ Task<MessageBodyView> GetMessageBodyFromMetadata(string messageId)
             return Task.FromResult(MessageBodyView.FromString(body, contentType, bodySize, string.Empty));
         }
 
-        public async Task<QueryResult<IList<KnownEndpointsView>>> QueryKnownEndpoints(CancellationToken cancellationToken)
-        {
-            var knownEndpointsView = knownEndpoints
-                .Select(x => new KnownEndpointsView
-                {
-                    Id = DeterministicGuid.MakeId(x.Name, x.HostId.ToString()),
-                    EndpointDetails = new EndpointDetails
-                    {
-                        Host = x.Host,
-                        HostId = x.HostId,
-                        Name = x.Name
-                    },
-                    HostDisplayName = x.Host
-                })
-                .ToList();
-
-            return await Task.FromResult(new QueryResult<IList<KnownEndpointsView>>(knownEndpointsView, new QueryStatsInfo(string.Empty, knownEndpointsView.Count)));
-        }
-
         public Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken)
         {
             var results = messageViews

src/ServiceControl.Audit.Persistence.InMemory/InMemoryAuditIngestionUnitOfWork.cs

@@ -5,7 +5,6 @@
     using System.Threading.Tasks;
     using Auditing.BodyStorage;
     using ServiceControl.Audit.Auditing;
-    using ServiceControl.Audit.Persistence.Monitoring;
     using ServiceControl.Audit.Persistence.UnitOfWork;
     using ServiceControl.SagaAudit;
 
@@ -16,12 +15,6 @@ class InMemoryAuditIngestionUnitOfWork(
     {
         public ValueTask DisposeAsync() => ValueTask.CompletedTask;
 
-        public Task RecordKnownEndpoint(KnownEndpoint knownEndpoint, CancellationToken cancellationToken)
-        {
-            dataStore.knownEndpoints.Add(knownEndpoint);
-            return Task.CompletedTask;
-        }
-
         public async Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory<byte> body, CancellationToken cancellationToken)
         {
             if (!body.IsEmpty)

src/ServiceControl.Audit.Persistence.RavenDB/RavenAuditDataStore.cs

@@ -8,12 +8,9 @@
     using Auditing.MessagesView;
     using Extensions;
     using Indexes;
-    using Monitoring;
     using Raven.Client.Documents;
     using ServiceControl.Audit.Auditing;
     using ServiceControl.Audit.Infrastructure;
-    using ServiceControl.Audit.Monitoring;
-    using ServiceControl.Audit.Persistence.Infrastructure;
     using ServiceControl.SagaAudit;
     using Transformers;
 
@@ -125,27 +122,6 @@ public async Task<MessageBodyView> GetMessageBody(string messageId, Cancellation
             );
         }
 
-        public async Task<QueryResult<IList<KnownEndpointsView>>> QueryKnownEndpoints(CancellationToken cancellationToken)
-        {
-            using var session = await sessionProvider.OpenSession(cancellationToken: cancellationToken);
-            var endpoints = await session.Advanced.LoadStartingWithAsync<KnownEndpoint>(KnownEndpoint.CollectionName, pageSize: 1024, token: cancellationToken);
-
-            var knownEndpoints = endpoints
-                .Select(x => new KnownEndpointsView
-                {
-                    Id = DeterministicGuid.MakeId(x.Name, x.HostId.ToString()),
-                    EndpointDetails = new EndpointDetails
-                    {
-                        Host = x.Host,
-                        HostId = x.HostId,
-                        Name = x.Name
-                    },
-                    HostDisplayName = x.Host
-                })
-                .ToList();
-
-            return new QueryResult<IList<KnownEndpointsView>>(knownEndpoints, new QueryStatsInfo(string.Empty, knownEndpoints.Count));
-        }
 
         public async Task<QueryResult<IList<AuditCount>>> QueryAuditCounts(string endpointName, CancellationToken cancellationToken)
         {

src/ServiceControl.Audit.Persistence.RavenDB/UnitOfWork/RavenAuditIngestionUnitOfWork.cs

@@ -6,7 +6,6 @@
     using System.Threading.Tasks;
     using Auditing;
     using Auditing.BodyStorage;
-    using Monitoring;
     using NServiceBus;
     using Persistence.UnitOfWork;
     using Raven.Client;
@@ -50,9 +49,6 @@ MetadataAsDictionary GetExpirationMetadata() =>
         public Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken)
             => bulkInsert.StoreAsync(sagaSnapshot, GetExpirationMetadata());
 
-        public Task RecordKnownEndpoint(KnownEndpoint knownEndpoint, CancellationToken cancellationToken)
-            => bulkInsert.StoreAsync(knownEndpoint, GetExpirationMetadata());
-
         public async ValueTask DisposeAsync()
         {
             await bulkInsert.DisposeAsync();

src/ServiceControl.Audit.Persistence.Tests.RavenDB/EmbeddedLifecycleTests.cs

@@ -4,6 +4,7 @@
     using System.IO;
     using System.Threading.Tasks;
     using NUnit.Framework;
+    using ServiceControl.Audit.Infrastructure;
     using ServiceControl.Audit.Persistence.RavenDB;
     using TestHelper;
 
@@ -35,7 +36,7 @@ public override async Task Setup()
         [Test]
         public async Task Verify_embedded_database()
         {
-            await DataStore.QueryKnownEndpoints(TestContext.CurrentContext.CancellationToken);
+            await DataStore.GetMessages(false, new PagingInfo(), new SortInfo("Id", "asc"), cancellationToken: TestContext.CurrentContext.CancellationToken);
 
             using (Assert.EnterMultipleScope())
             {

src/ServiceControl.Audit.Persistence.Tests.RavenDB/RetentionTests.cs

@@ -4,7 +4,6 @@
     using System.Collections.Generic;
     using System.Threading.Tasks;
     using Auditing;
-    using Monitoring;
     using NServiceBus;
     using NUnit.Framework;
     using SagaAudit;
@@ -44,36 +43,6 @@ public async Task AuditMessageRetention()
             }
         }
 
-        [Test]
-        public async Task KnownEndpointRetention()
-        {
-            var knownEndpoint = new KnownEndpoint()
-            {
-                Host = "Myself",
-                HostId = Guid.NewGuid(),
-                Id = "KnownEndpoints/1234123",
-                LastSeen = DateTime.UtcNow,
-                Name = "Wazowsky"
-            };
-
-            await IngestKnownEndpoints(
-                knownEndpoint
-            );
-
-            var queryResultBeforeExpiration = await DataStore.QueryKnownEndpoints(TestContext.CurrentContext.CancellationToken);
-
-            await Task.Delay(4000);
-
-            var queryResultAfterExpiration = await DataStore.QueryKnownEndpoints(TestContext.CurrentContext.CancellationToken);
-
-            Assert.That(queryResultBeforeExpiration.Results, Has.Count.EqualTo(1));
-            using (Assert.EnterMultipleScope())
-            {
-                Assert.That(queryResultBeforeExpiration.Results[0].EndpointDetails.Name, Is.EqualTo("Wazowsky"));
-                Assert.That(queryResultAfterExpiration.Results.Count, Is.EqualTo(0));
-            }
-        }
-
         [Test]
         public async Task SagaSnapshotRetention()
         {
@@ -147,17 +116,6 @@ async Task IngestProcessedMessagesAudits(params ProcessedMessage[] processedMess
             await configuration.CompleteDBOperation();
         }
 
-        async Task IngestKnownEndpoints(params KnownEndpoint[] knownEndpoints)
-        {
-            var unitOfWork = await StartAuditUnitOfWork(knownEndpoints.Length);
-            foreach (var knownEndpoint in knownEndpoints)
-            {
-                await unitOfWork.RecordKnownEndpoint(knownEndpoint);
-            }
-            await unitOfWork.DisposeAsync();
-            await configuration.CompleteDBOperation();
-        }
-
         async Task IngestSagaAudits(params SagaSnapshot[] snapshots)
         {
             var unitOfWork = await StartAuditUnitOfWork(snapshots.Length);

src/ServiceControl.Audit.Persistence.Tests/KnownEndpointsTests.cs

@@ -12,7 +12,6 @@
 
     public interface IAuditDataStore
     {
-        Task<QueryResult<IList<KnownEndpointsView>>> QueryKnownEndpoints(CancellationToken cancellationToken);
         Task<QueryResult<SagaHistory>> QuerySagaHistoryById(Guid input, CancellationToken cancellationToken);
         Task<QueryResult<IList<MessagesView>>> GetMessages(bool includeSystemMessages, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default);
         Task<QueryResult<IList<MessagesView>>> QueryMessages(string searchParam, PagingInfo pagingInfo, SortInfo sortInfo, DateTimeRange timeSentRange = null, CancellationToken cancellationToken = default);

src/ServiceControl.Audit.Persistence/IAuditDataStore.cs

@@ -4,13 +4,11 @@
     using System.Threading;
     using System.Threading.Tasks;
     using Auditing;
-    using Monitoring;
     using ServiceControl.SagaAudit;
 
     public interface IAuditIngestionUnitOfWork : IAsyncDisposable
     {
         Task RecordProcessedMessage(ProcessedMessage processedMessage, ReadOnlyMemory<byte> body = default, CancellationToken cancellationToken = default);
         Task RecordSagaSnapshot(SagaSnapshot sagaSnapshot, CancellationToken cancellationToken = default);
-        Task RecordKnownEndpoint(KnownEndpoint knownEndpoint, CancellationToken cancellationToken = default);
     }
 }