Merged otel-metrics

ramonsmits · ramonsmits · commit 58043ae9ee1b · 2025-01-31T14:41:04.000+01:00
diff --git a/.github/workflows/build-windows.yml b/.github/workflows/build-windows.yml
@@ -15,7 +15,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Setup .NET SDK
-        uses: actions/setup-dotnet@v4.2.0
+        uses: actions/setup-dotnet@v4.3.0
         with:
           dotnet-version: 8.0.x
       - name: Download RavenDB Server
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -36,7 +36,7 @@ jobs:
         with:
           fetch-depth: 0
       - name: Setup .NET SDK
-        uses: actions/setup-dotnet@v4.2.0
+        uses: actions/setup-dotnet@v4.3.0
         with:
           dotnet-version: 8.0.x
       - name: Download RavenDB Server
diff --git a/src/Directory.Packages.props b/src/Directory.Packages.props
@@ -8,6 +8,7 @@
     <PackageVersion Include="AWSSDK.CloudWatch" Version="3.7.402.21" />
     <PackageVersion Include="AWSSDK.SecurityToken" Version="3.7.401.22" />
     <PackageVersion Include="Azure.Identity" Version="1.13.1" />
+    <PackageVersion Include="Azure.Monitor.OpenTelemetry.Exporter" Version="1.3.0" />
     <PackageVersion Include="Azure.Monitor.Query" Version="1.6.0" />
     <PackageVersion Include="Azure.ResourceManager.ServiceBus" Version="1.0.1" />
     <PackageVersion Include="ByteSize" Version="2.1.2" />
@@ -50,6 +51,9 @@
     <PackageVersion Include="NUnit" Version="4.3.2" />
     <PackageVersion Include="NUnit.Analyzers" Version="4.6.0" />
     <PackageVersion Include="NUnit3TestAdapter" Version="4.6.0" />
+    <PackageVersion Include="OpenTelemetry.Exporter.Console" Version="1.9.0" />
+    <PackageVersion Include="OpenTelemetry.Exporter.OpenTelemetryProtocol" Version="1.9.0" />
+    <PackageVersion Include="OpenTelemetry.Extensions.Hosting" Version="1.9.0" />
     <PackageVersion Include="Particular.Approvals" Version="2.0.1" />
     <PackageVersion Include="Particular.Licensing.Sources" Version="6.0.0" />
     <PackageVersion Include="Particular.LicensingComponent.Report" Version="1.0.0" />
@@ -87,4 +91,4 @@
     <GlobalPackageReference Include="Microsoft.Build.CopyOnWrite" Version="1.0.334" />
     <GlobalPackageReference Include="Particular.Packaging" Version="4.2.0" />
   </ItemGroup>
-</Project>
+</Project>
diff --git a/src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs b/src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs
@@ -51,6 +51,7 @@ async Task InitializeServiceControl(ScenarioContext context)
                 TransportConnectionString = transportToUse.ConnectionString,
                 MaximumConcurrencyLevel = 2,
                 ServiceControlQueueAddress = "SHOULDNOTBEUSED",
+                OtelMetricsUrl = "http://localhost:4317",
                 MessageFilter = messageContext =>
                 {
                     var id = messageContext.NativeMessageId;
diff --git a/src/ServiceControl.Audit/App.config b/src/ServiceControl.Audit/App.config
@@ -9,7 +9,7 @@ These settings are only here so that we can debug ServiceControl while developin
     <add key="ServiceControl.Audit/HostName" value="localhost" />
     <add key="ServiceControl.Audit/DatabaseMaintenancePort" value="44445" />
     <add key="ServiceControl.Audit/LogLevel" value="Debug" />
-    <add key="ServiceControl.Audit/PrintMetrics" value="True" />
+    <add key="ServiceControl.Audit/OtelMetricsUrl" value="http://localhost:4317" />
 
     <!-- DEVS - Pick a transport to run Auditing instance on -->
     <add key="ServiceControl.Audit/TransportType" value="LearningTransport" />
diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestion.cs b/src/ServiceControl.Audit/Auditing/AuditIngestion.cs
@@ -3,6 +3,7 @@
     using System;
     using System.Collections.Generic;
     using System.Diagnostics;
+    using System.Diagnostics.Metrics;
     using System.Threading;
     using System.Threading.Channels;
     using System.Threading.Tasks;
@@ -14,18 +15,14 @@
     using Persistence;
     using Persistence.UnitOfWork;
     using ServiceControl.Infrastructure;
-    using ServiceControl.Infrastructure.Metrics;
     using Transports;
 
     class AuditIngestion : IHostedService
     {
-        static readonly long FrequencyInMilliseconds = Stopwatch.Frequency / 1000;
-
         public AuditIngestion(
             Settings settings,
             ITransportCustomization transportCustomization,
             TransportSettings transportSettings,
-            Metrics metrics,
             IFailedAuditStorage failedImportsStorage,
             AuditIngestionCustomCheck.State ingestionState,
             AuditIngestor auditIngestor,
@@ -40,10 +37,6 @@ public AuditIngestion(
             this.settings = settings;
             this.applicationLifetime = applicationLifetime;
 
-            batchSizeMeter = metrics.GetMeter("Audit ingestion - batch size");
-            batchDurationMeter = metrics.GetMeter("Audit ingestion - batch processing duration", FrequencyInMilliseconds);
-            receivedMeter = metrics.GetCounter("Audit ingestion - received");
-
             if (!transportSettings.MaxConcurrency.HasValue)
             {
                 throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
@@ -102,7 +95,6 @@ async Task EnsureStarted(CancellationToken cancellationToken = default)
                         await stoppable.StopReceive(cancellationToken);
                         logger.Info("Shutting down due to failed persistence health check. Infrastructure shut down completed");
                     }
-
                     return;
                 }
 
@@ -199,7 +191,7 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati
             var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
             messageContext.SetTaskCompletionSource(taskCompletionSource);
 
-            receivedMeter.Mark();
+            receivedMeter.Add(1);
 
             await channel.Writer.WriteAsync(messageContext, cancellationToken);
             await taskCompletionSource.Task;
@@ -222,11 +214,11 @@ async Task Loop()
                             contexts.Add(context);
                         }
 
-                        batchSizeMeter.Mark(contexts.Count);
-                        using (batchDurationMeter.Measure())
-                        {
-                            await auditIngestor.Ingest(contexts);
-                        }
+                        batchSizeMeter.Record(contexts.Count);
+                        var sw = Stopwatch.StartNew();
+
+                        await auditIngestor.Ingest(contexts);
+                        batchDurationMeter.Record(sw.ElapsedMilliseconds);
                     }
                     catch (OperationCanceledException e)
                     {
@@ -274,9 +266,9 @@ async Task Loop()
         readonly IAuditIngestionUnitOfWorkFactory unitOfWorkFactory;
         readonly Settings settings;
         readonly Channel<MessageContext> channel;
-        readonly Meter batchSizeMeter;
-        readonly Meter batchDurationMeter;
-        readonly Counter receivedMeter;
+        readonly Histogram<long> batchSizeMeter = AuditMetrics.Meter.CreateHistogram<long>($"{AuditMetrics.Prefix}.batch_size");
+        readonly Histogram<double> batchDurationMeter = AuditMetrics.Meter.CreateHistogram<double>($"{AuditMetrics.Prefix}.batch_duration_ms");
+        readonly Counter<long> receivedMeter = AuditMetrics.Meter.CreateCounter<long>($"{AuditMetrics.Prefix}.received");
         readonly Watchdog watchdog;
         readonly Task ingestionWorker;
         readonly IHostApplicationLifetime applicationLifetime;
diff --git a/src/ServiceControl.Audit/Auditing/AuditIngestor.cs b/src/ServiceControl.Audit/Auditing/AuditIngestor.cs
@@ -14,13 +14,11 @@
     using Persistence.UnitOfWork;
     using Recoverability;
     using SagaAudit;
-    using ServiceControl.Infrastructure.Metrics;
     using ServiceControl.Transports;
 
     public class AuditIngestor
     {
         public AuditIngestor(
-            Metrics metrics,
             Settings settings,
             IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
             EndpointInstanceMonitoring endpointInstanceMonitoring,
@@ -32,13 +30,6 @@ ITransportCustomization transportCustomization
         {
             this.settings = settings;
             this.messageDispatcher = messageDispatcher;
-
-            var ingestedAuditMeter = metrics.GetCounter("Audit ingestion - ingested audit");
-            var ingestedSagaAuditMeter = metrics.GetCounter("Audit ingestion - ingested saga audit");
-            var auditBulkInsertDurationMeter = metrics.GetMeter("Audit ingestion - audit bulk insert duration", FrequencyInMilliseconds);
-            var sagaAuditBulkInsertDurationMeter = metrics.GetMeter("Audit ingestion - saga audit bulk insert duration", FrequencyInMilliseconds);
-            var bulkInsertCommitDurationMeter = metrics.GetMeter("Audit ingestion - bulk insert commit duration", FrequencyInMilliseconds);
-
             var enrichers = new IEnrichImportedAuditMessages[]
             {
                 new MessageTypeEnricher(),
@@ -51,7 +42,7 @@ ITransportCustomization transportCustomization
 
             logQueueAddress = transportCustomization.ToTransportQualifiedQueueName(settings.AuditLogQueue);
 
-            auditPersister = new AuditPersister(unitOfWorkFactory, enrichers, ingestedAuditMeter, ingestedSagaAuditMeter, auditBulkInsertDurationMeter, sagaAuditBulkInsertDurationMeter, bulkInsertCommitDurationMeter, messageSession, messageDispatcher);
+            auditPersister = new AuditPersister(unitOfWorkFactory, enrichers, messageSession, messageDispatcher);
         }
 
         public async Task Ingest(List<MessageContext> contexts)
@@ -159,7 +150,6 @@ public async Task VerifyCanReachForwardingAddress()
         readonly Lazy<IMessageDispatcher> messageDispatcher;
         readonly string logQueueAddress;
 
-        static readonly long FrequencyInMilliseconds = Stopwatch.Frequency / 1000;
         static readonly ILog Log = LogManager.GetLogger<AuditIngestor>();
     }
 }
diff --git a/src/ServiceControl.Audit/Auditing/AuditMetrics.cs b/src/ServiceControl.Audit/Auditing/AuditMetrics.cs
@@ -0,0 +1,9 @@
+namespace ServiceControl.Audit.Auditing;
+
+using System.Diagnostics.Metrics;
+
+static class AuditMetrics
+{
+    public static readonly Meter Meter = new("ServiceControl", "0.1.0");
+    public static readonly string Prefix = "particular.servicecontrol.audit";
+}
diff --git a/src/ServiceControl.Audit/Auditing/AuditPersister.cs b/src/ServiceControl.Audit/Auditing/AuditPersister.cs
@@ -3,6 +3,7 @@
     using System;
     using System.Collections.Generic;
     using System.Diagnostics;
+    using System.Diagnostics.Metrics;
     using System.Text.Json;
     using System.Threading.Tasks;
     using Infrastructure;
@@ -15,36 +16,20 @@
     using ServiceControl.Audit.Persistence.Monitoring;
     using ServiceControl.EndpointPlugin.Messages.SagaState;
     using ServiceControl.Infrastructure;
-    using ServiceControl.Infrastructure.Metrics;
     using ServiceControl.SagaAudit;
 
-    class AuditPersister
+    class AuditPersister(IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
+        IEnrichImportedAuditMessages[] enrichers,
+        IMessageSession messageSession,
+        Lazy<IMessageDispatcher> messageDispatcher)
     {
-        public AuditPersister(IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
-            IEnrichImportedAuditMessages[] enrichers,
-            Counter ingestedAuditMeter, Counter ingestedSagaAuditMeter, Meter auditBulkInsertDurationMeter,
-            Meter sagaAuditBulkInsertDurationMeter, Meter bulkInsertCommitDurationMeter, IMessageSession messageSession,
-            Lazy<IMessageDispatcher> messageDispatcher)
-        {
-            this.unitOfWorkFactory = unitOfWorkFactory;
-            this.enrichers = enrichers;
-
-            this.ingestedAuditMeter = ingestedAuditMeter;
-            this.ingestedSagaAuditMeter = ingestedSagaAuditMeter;
-            this.auditBulkInsertDurationMeter = auditBulkInsertDurationMeter;
-            this.sagaAuditBulkInsertDurationMeter = sagaAuditBulkInsertDurationMeter;
-            this.bulkInsertCommitDurationMeter = bulkInsertCommitDurationMeter;
-            this.messageSession = messageSession;
-            this.messageDispatcher = messageDispatcher;
-        }
-
         public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageContext> contexts)
         {
             var stopwatch = Stopwatch.StartNew();
 
             if (Logger.IsDebugEnabled)
             {
-                Logger.DebugFormat("Batch size {0}", contexts.Count);
+                Logger.Debug($"Batch size {contexts.Count}");
             }
 
             var storedContexts = new List<MessageContext>(contexts.Count);
@@ -86,23 +71,25 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
 
                         Logger.Debug("Adding audit message for bulk storage");
 
-                        using (auditBulkInsertDurationMeter.Measure())
-                        {
-                            await unitOfWork.RecordProcessedMessage(processedMessage, context.Body);
-                        }
+                        var auditSw = Stopwatch.StartNew();
+                        await unitOfWork.RecordProcessedMessage(processedMessage, context.Body);
+                        auditSw.Stop();
+
+                        auditBulkInsertDurationMeter.Record(auditSw.ElapsedMilliseconds);
 
-                        ingestedAuditMeter.Mark();
+                        ingestedAuditMeter.Add(1);
                     }
                     else if (context.Extensions.TryGet(out SagaSnapshot sagaSnapshot))
                     {
                         Logger.Debug("Adding SagaSnapshot message for bulk storage");
 
-                        using (sagaAuditBulkInsertDurationMeter.Measure())
-                        {
-                            await unitOfWork.RecordSagaSnapshot(sagaSnapshot);
-                        }
+                        var sagaSw = Stopwatch.StartNew();
+                        await unitOfWork.RecordSagaSnapshot(sagaSnapshot);
+                        sagaSw.Stop();
+
+                        sagaAuditBulkInsertDurationMeter.Record(sagaSw.ElapsedMilliseconds);
 
-                        ingestedSagaAuditMeter.Mark();
+                        ingestedSagaAuditMeter.Add(1);
                     }
 
                     storedContexts.Add(context);
@@ -131,10 +118,10 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
                     try
                     {
                         // this can throw even though dispose is never supposed to throw
-                        using (bulkInsertCommitDurationMeter.Measure())
-                        {
-                            await unitOfWork.DisposeAsync();
-                        }
+                        var commitSw = Stopwatch.StartNew();
+                        await unitOfWork.DisposeAsync();
+                        commitSw.Stop();
+                        bulkInsertCommitDurationMeter.Record(commitSw.ElapsedMilliseconds);
                     }
                     catch (Exception e)
                     {
@@ -284,16 +271,12 @@ await messageDispatcher.Value.Dispatch(new TransportOperations(messagesToEmit.To
             }
         }
 
-        readonly Counter ingestedAuditMeter;
-        readonly Counter ingestedSagaAuditMeter;
-        readonly Meter auditBulkInsertDurationMeter;
-        readonly Meter sagaAuditBulkInsertDurationMeter;
-        readonly Meter bulkInsertCommitDurationMeter;
-        readonly IMessageSession messageSession;
-        readonly Lazy<IMessageDispatcher> messageDispatcher;
+        readonly Counter<long> ingestedAuditMeter = AuditMetrics.Meter.CreateCounter<long>($"{AuditMetrics.Prefix}.ingested_audit_messages"); // metrics.GetCounter("Audit ingestion - ingested audit");
+        readonly Counter<long> ingestedSagaAuditMeter = AuditMetrics.Meter.CreateCounter<long>($"{AuditMetrics.Prefix}.ingested_saga_audits"); // metrics.GetCounter("Audit ingestion - ingested audit");
+        readonly Histogram<double> auditBulkInsertDurationMeter = AuditMetrics.Meter.CreateHistogram<double>($"{AuditMetrics.Prefix}.audit_bulk_insert_duration_ms"); // metrics.GetCounter("Audit ingestion - ingested audit");
+        readonly Histogram<double> sagaAuditBulkInsertDurationMeter = AuditMetrics.Meter.CreateHistogram<double>($"{AuditMetrics.Prefix}.saga_bulk_insert_duration_ms"); // metrics.GetCounter("Audit ingestion - ingested audit");
+        readonly Histogram<double> bulkInsertCommitDurationMeter = AuditMetrics.Meter.CreateHistogram<double>($"{AuditMetrics.Prefix}.audit_commit_duration_ms"); // metrics.GetCounter("Audit ingestion - ingested audit");
 
-        readonly IEnrichImportedAuditMessages[] enrichers;
-        readonly IAuditIngestionUnitOfWorkFactory unitOfWorkFactory;
         static readonly ILog Logger = LogManager.GetLogger<AuditPersister>();
     }
 }
diff --git a/src/ServiceControl.Audit/HostApplicationBuilderExtensions.cs b/src/ServiceControl.Audit/HostApplicationBuilderExtensions.cs
@@ -5,8 +5,8 @@ namespace ServiceControl.Audit;
 using System.Threading;
 using System.Threading.Tasks;
 using Auditing;
+using Azure.Monitor.OpenTelemetry.Exporter;
 using Infrastructure;
-using Infrastructure.Metrics;
 using Infrastructure.Settings;
 using Microsoft.AspNetCore.HttpLogging;
 using Microsoft.Extensions.DependencyInjection;
@@ -20,6 +20,8 @@ namespace ServiceControl.Audit;
 using NServiceBus.Transport;
 using Persistence;
 using Transports;
+using OpenTelemetry.Metrics;
+using OpenTelemetry.Resources;
 
 static class HostApplicationBuilderExtensions
 {
@@ -61,13 +63,38 @@ public static void AddServiceControlAudit(this IHostApplicationBuilder builder,
         // directly and to make things more complex of course the order of registration still matters ;)
         services.AddSingleton(provider => new Lazy<IMessageDispatcher>(provider.GetRequiredService<IMessageDispatcher>));
 
-        services.AddMetrics(settings.PrintMetrics);
-
         services.AddPersistence(persistenceSettings, persistenceConfiguration);
 
         NServiceBusFactory.Configure(settings, transportCustomization, transportSettings, onCriticalError, configuration);
         builder.UseNServiceBus(configuration);
 
+        if (!string.IsNullOrEmpty(settings.OtelMetricsUrl))
+        {
+            builder.Services.AddOpenTelemetry()
+                .ConfigureResource(b => b.AddService(serviceName: settings.InstanceName))
+                .WithMetrics(b =>
+                {
+                    b.AddMeter("ServiceControl");
+
+                    if (Uri.TryCreate(settings.OtelMetricsUrl, UriKind.Absolute, out var uri))
+                    {
+                        b.AddOtlpExporter(e =>
+                        {
+                            e.Endpoint = uri;
+                        });
+                    }
+                    else
+                    {
+                        b.AddAzureMonitorMetricExporter(o =>
+                        {
+                            o.ConnectionString = settings.OtelMetricsUrl;
+                        });
+                    }
+
+                    b.AddConsoleExporter();
+                });
+        }
+
         // Configure after the NServiceBus hosted service to ensure NServiceBus is already started
         if (settings.IngestAuditMessages)
         {
diff --git a/src/ServiceControl.Audit/Infrastructure/Metrics/MetricsReporterHostedService.cs b/src/ServiceControl.Audit/Infrastructure/Metrics/MetricsReporterHostedService.cs
diff --git a/src/ServiceControl.Audit/Infrastructure/Metrics/MetricsServiceCollectionExtensions.cs b/src/ServiceControl.Audit/Infrastructure/Metrics/MetricsServiceCollectionExtensions.cs
diff --git a/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs b/src/ServiceControl.Audit/Infrastructure/Settings/Settings.cs
diff --git a/src/ServiceControl.Audit/ServiceControl.Audit.csproj b/src/ServiceControl.Audit/ServiceControl.Audit.csproj

Original file line number	Diff line number	Diff line change
`@@ -51,6 +51,7 @@ async Task InitializeServiceControl(ScenarioContext context)`
`51`	`51`	`TransportConnectionString = transportToUse.ConnectionString,`
`52`	`52`	`MaximumConcurrencyLevel = 2,`
`53`	`53`	`ServiceControlQueueAddress = "SHOULDNOTBEUSED",`
	`54`	`+ OtelMetricsUrl = "http://localhost:4317",`
`54`	`55`	`MessageFilter = messageContext =>`
`55`	`56`	`{`
`56`	`57`	`var id = messageContext.NativeMessageId;`