[Spike] Run audit ingestion in the primary instance

src/Particular.LicensingComponent.UnitTests/AuditQuery_Tests.cs

@@ -10,6 +10,7 @@
 using NUnit.Framework;
 using Particular.Approvals;
 using Particular.LicensingComponent.AuditThroughput;
+using Particular.LicensingComponent.Contracts;
 using Particular.LicensingComponent.UnitTests.Infrastructure;
 using ServiceControl.Api;
 using ServiceControl.Api.Contracts;
@@ -73,6 +74,43 @@ public async Task Should_return_audit_remotes()
         }
     }
 
+    [Test]
+    public async Task Should_return_the_local_audit_source_alongside_the_remotes()
+    {
+        //Arrange
+        var auditQuery = new AuditQuery(NullLogger<AuditQuery>.Instance, new FakeEndpointApi(), new FakeAuditCountApi(),
+            new ConfigurationApi_ReturningOneValidAuditConfig(), new LocalAuditSource_ForThisInstance());
+
+        //Act
+        var remotes = await auditQuery.GetAuditRemotes();
+
+        //Assert
+        Assert.That(remotes, Has.Count.EqualTo(2), "The local audit source and the remote should both be reported");
+
+        var local = remotes.Single(remote => remote.ApiUri == "http://localhost:33333/api/");
+
+        using (Assert.EnterMultipleScope())
+        {
+            Assert.That(local.Queues, Does.Contain("audit"), "the local audit queue must be recognised as a platform endpoint");
+            Assert.That(local.Queues, Does.Contain("audit.log"));
+            Assert.That(local.Transport, Is.EqualTo("LearningTransport"), "the report's audit service metadata is built from this");
+        }
+    }
+
+    [Test]
+    public async Task Should_not_report_a_local_audit_source_that_is_disabled()
+    {
+        //Arrange
+        var auditQuery = new AuditQuery(NullLogger<AuditQuery>.Instance, new FakeEndpointApi(), new FakeAuditCountApi(),
+            new ConfigurationApi_ReturningOneValidAuditConfig(), new LocalAuditSource_Disabled());
+
+        //Act
+        var remotes = await auditQuery.GetAuditRemotes();
+
+        //Assert
+        Assert.That(remotes, Has.Count.EqualTo(1));
+    }
+
     [Test]
     public async Task Should_return_successful_audit_connection_if_instances_exist_and_are_online()
     {
@@ -203,6 +241,29 @@ public Task<List<Endpoint>> GetEndpoints(CancellationToken cancellationToken = d
 
     }
 
+    class LocalAuditSource_ForThisInstance : ILocalAuditSource
+    {
+        public bool Enabled => true;
+
+        public RemoteInstanceInformation Describe() => new()
+        {
+            ApiUri = "http://localhost:33333/api/",
+            VersionString = "6.0.0",
+            SemanticVersion = new NuGet.Versioning.SemanticVersion(6, 0, 0),
+            Status = "online",
+            Retention = TimeSpan.FromDays(10),
+            Queues = ["audit", "audit.log"],
+            Transport = "LearningTransport"
+        };
+    }
+
+    class LocalAuditSource_Disabled : ILocalAuditSource
+    {
+        public bool Enabled => false;
+
+        public RemoteInstanceInformation Describe() => throw new InvalidOperationException("Describe must not be called when the source is disabled.");
+    }
+
     class AuditCountApi_ReturningThreeAuditCounts : IAuditCountApi
     {
         public async Task<IList<AuditCount>> GetEndpointAuditCounts(string endpoint, CancellationToken cancellationToken = default)

src/Particular.LicensingComponent/AuditThroughput/AuditQuery.cs

@@ -7,7 +7,7 @@
     using ServiceControl.Api;
     using AuditCount = Contracts.AuditCount;
 
-    public class AuditQuery(ILogger<AuditQuery> logger, IEndpointsApi endpointsApi, IAuditCountApi auditCountApi, IConfigurationApi configurationApi) : IAuditQuery
+    public class AuditQuery(ILogger<AuditQuery> logger, IEndpointsApi endpointsApi, IAuditCountApi auditCountApi, IConfigurationApi configurationApi, ILocalAuditSource? localAuditSource = null) : IAuditQuery
     {
         // Customers are expected to run at least version 4.29 for their Audit instances
         public SemanticVersion MinAuditCountsVersion => new(4, 29, 0);
@@ -45,6 +45,11 @@ public async Task<List<RemoteInstanceInformation>> GetAuditRemotes(CancellationT
                 var remotes = await configurationApi.GetRemoteConfigs(cancellationToken);
                 var remotesInfo = new List<RemoteInstanceInformation>();
 
+                if (localAuditSource is { Enabled: true })
+                {
+                    remotesInfo.Add(localAuditSource.Describe());
+                }
+
                 if (remotes.Any())
                 {
                     List<string> queues = [];

src/Particular.LicensingComponent/AuditThroughput/ILocalAuditSource.cs

@@ -0,0 +1,15 @@
+namespace Particular.LicensingComponent.AuditThroughput;
+
+using Particular.LicensingComponent.Contracts;
+
+/// <summary>
+/// Audit throughput collection is driven entirely by audit remotes. A primary that holds audit data
+/// itself has no remote to describe it, so without this its own audit queues are counted as customer
+/// endpoints and the audit service metadata in the licensing report is blank.
+/// </summary>
+public interface ILocalAuditSource
+{
+    bool Enabled { get; }
+
+    RemoteInstanceInformation Describe();
+}

src/Particular.LicensingComponent/LicensingComponentHostBuilderExtensions.cs

@@ -1,4 +1,4 @@
-namespace Particular.LicensingComponent;
+namespace Particular.LicensingComponent;
 
 using AuditThroughput;
 using BrokerThroughput;

src/ServiceControl.AcceptanceTests.RavenDB/ServiceControl.AcceptanceTests.RavenDB.csproj

@@ -42,6 +42,10 @@
 
     <!-- RavenDB is the one persister that supports maintenance mode; StartupModeTests covers it. -->
     <Compile Remove="..\ServiceControl.AcceptanceTests\MaintenanceModeTests.cs" />
+
+    <Compile Remove="..\ServiceControl.AcceptanceTests\Auditing\When_composing_audit_ingestion_in_the_primary.cs" />
+    <Compile Remove="..\ServiceControl.AcceptanceTests\Auditing\When_hosting_an_audit_instance.cs" />
+    <Compile Remove="..\ServiceControl.AcceptanceTests\Auditing\When_hosting_audit_ingestion_only.cs" />
   </ItemGroup>
 
   <ItemGroup>

src/ServiceControl.AcceptanceTests/Auditing/When_composing_audit_ingestion_in_the_primary.cs

@@ -0,0 +1,162 @@
+namespace ServiceControl.AcceptanceTests.Auditing
+{
+    using System;
+    using System.IO;
+    using System.Linq;
+    using System.Runtime.Loader;
+    using System.Threading.Tasks;
+    using Microsoft.AspNetCore.Builder;
+    using Microsoft.Extensions.DependencyInjection;
+    using Microsoft.Extensions.Hosting;
+    using Microsoft.Extensions.Logging;
+    using NServiceBus;
+    using NUnit.Framework;
+    using Particular.LicensingComponent.AuditThroughput;
+    using Particular.ServiceControl;
+    using ServiceBus.Management.Infrastructure.Settings;
+    using ServiceControl.Auditing;
+    using ServiceControl.Connection;
+    using ServiceControl.Infrastructure;
+    using ServiceControl.Infrastructure.WebApi;
+    using ServiceControl.Persistence;
+    using ServiceControl.SagaAudit;
+
+    class When_composing_audit_ingestion_in_the_primary : AcceptanceTest
+    {
+        [Test]
+        public async Task Should_host_the_audit_runtime()
+        {
+            var (app, services) = await BuildHost();
+
+            try
+            {
+                using (Assert.EnterMultipleScope())
+                {
+                    Assert.That(HostsAuditIngestion(services), Is.True);
+                    Assert.That(app.Services.GetService<AuditIngestor>(), Is.Not.Null);
+                    Assert.That(app.Services.GetService<ImportFailedAudits>(), Is.Not.Null);
+                    Assert.That(app.Services.GetService<IFailedAuditImportDataStore>(), Is.Not.Null);
+                    Assert.That(app.Services.GetService<IAuditCountsDataStore>(), Is.Not.Null);
+                    Assert.That(app.Services.GetService<ISagaHistoryDataStore>(), Is.Not.Null);
+
+                    Assert.That(app.Services.GetService<ILocalAuditSource>(), Is.Not.Null,
+                        "without it the local audit queues are counted as customer endpoints in the licensing report");
+                    Assert.That(services.Any(descriptor =>
+                            descriptor.ServiceType == typeof(IProvidePlatformConnectionDetails)
+                            && descriptor.ImplementationType == typeof(AuditPlatformConnectionDetailsProvider)),
+                        Is.True,
+                        "/api/connection must still tell endpoints where to send audit and saga data");
+                }
+            }
+            finally
+            {
+                await app.DisposeAsync();
+            }
+        }
+
+        [Test]
+        public async Task Should_keep_every_audit_capability_but_the_receiver_when_ingestion_is_disabled()
+        {
+            var (app, services) = await BuildHost(settings => settings.IngestAuditMessages = false);
+
+            try
+            {
+                using (Assert.EnterMultipleScope())
+                {
+                    Assert.That(HostsAuditIngestion(services), Is.False,
+                        "the receiver is the only thing the setting turns off, because other processes may still be ingesting");
+                    Assert.That(app.Services.GetService<AuditIngestor>(), Is.Not.Null);
+                    Assert.That(app.Services.GetService<IAuditCountsDataStore>(), Is.Not.Null);
+                }
+            }
+            finally
+            {
+                await app.DisposeAsync();
+            }
+        }
+
+        [Test]
+        public async Task Should_stand_aside_from_audit_entirely_when_the_audit_data_is_remote()
+        {
+            var (app, services) = await BuildHost(settings =>
+            {
+                settings.AuditDataLocation = AuditDataLocation.Remote;
+                settings.RemoteInstances = [new RemoteInstanceSetting("http://localhost:44444/api")];
+            });
+
+            try
+            {
+                using (Assert.EnterMultipleScope())
+                {
+                    Assert.That(HostsAuditIngestion(services), Is.False, "the audit host ingests, not this primary");
+                    Assert.That(app.Services.GetService<AuditIngestor>(), Is.Null);
+                    Assert.That(app.Services.GetService<ILocalAuditSource>(), Is.Null, "licensing learns about audit from the remote");
+                    Assert.That(app.Services.GetRequiredService<IAuditCountsDataStore>(), Is.TypeOf<EmptyAuditCountsDataStore>(),
+                        "the local source stands aside so the scatter gather treats this instance as a non-participant");
+                    Assert.That(app.Services.GetRequiredService<ISagaHistoryDataStore>(), Is.TypeOf<EmptySagaHistoryDataStore>());
+                    Assert.That(app.Services.GetService<GetSagaByIdApi>(), Is.Not.Null, "the audit routes still answer, from the remote");
+                }
+            }
+            finally
+            {
+                await app.DisposeAsync();
+            }
+        }
+
+        [Test]
+        public async Task Should_refuse_local_audit_ingestion_alongside_remote_instances()
+        {
+            var exception = Assert.ThrowsAsync<Exception>(() => BuildHost(settings =>
+                settings.RemoteInstances = [new RemoteInstanceSetting("http://localhost:44444/api")]));
+
+            Assert.That(exception.Message, Does.Contain("AuditDataLocation"));
+        }
+
+        // The registrations are inspected rather than resolved. A normal primary hosts an NServiceBus
+        // endpoint, and constructing every hosted service without starting it fails inside the
+        // transport's receive component.
+        static bool HostsAuditIngestion(IServiceCollection services) =>
+            services.Any(descriptor =>
+                descriptor.ServiceType == typeof(IHostedService) && descriptor.ImplementationType == typeof(AuditIngestion));
+
+        async Task<(WebApplication App, IServiceCollection Services)> BuildHost(Action<Settings> customize = null)
+        {
+            var settings = await CreateSettings();
+
+            customize?.Invoke(settings);
+
+            var endpointConfiguration = new EndpointConfiguration(settings.InstanceName);
+            endpointConfiguration.AssemblyScanner().Disable = true;
+
+            var hostBuilder = WebApplication.CreateBuilder();
+            hostBuilder.AddServiceControl(settings, endpointConfiguration);
+            hostBuilder.AddServiceControlApi(settings.CorsSettings);
+
+            return (hostBuilder.Build(), hostBuilder.Services);
+        }
+
+        async Task<Settings> CreateSettings()
+        {
+            var settings = new Settings(TransportIntegration.TypeName, StorageConfiguration.PersistenceType,
+                CreateLoggingSettings(), forwardErrorMessages: false, errorRetentionPeriod: TimeSpan.FromDays(10))
+            {
+                InstanceName = $"AuditComposition.{Guid.NewGuid():n}",
+                TransportConnectionString = TransportIntegration.ConnectionString,
+                MaximumConcurrencyLevel = 2,
+                DisableHealthChecks = true,
+                AssemblyLoadContextResolver = static _ => AssemblyLoadContext.Default
+            };
+
+            await StorageConfiguration.CustomizeSettings(settings);
+
+            return settings;
+        }
+
+        static LoggingSettings CreateLoggingSettings()
+        {
+            var logPath = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName());
+            Directory.CreateDirectory(logPath);
+            return new LoggingSettings(Settings.SettingsRootNamespace, defaultLevel: LogLevel.Debug, logPath: logPath);
+        }
+    }
+}