Emit metrics via otel instead of custom format

Author: andreasohlund

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>

src/ServiceControl.Audit.AcceptanceTests/TestSupport/ServiceControlComponentRunner.cs

@@ -51,17 +51,19 @@ async Task InitializeServiceControl(ScenarioContext context)
                 TransportConnectionString = transportToUse.ConnectionString,
                 MaximumConcurrencyLevel = 2,
                 ServiceControlQueueAddress = "SHOULDNOTBEUSED",
+                OtelMetricsUrl = "http://localhost:4317",
                 MessageFilter = messageContext =>
                 {
                     var id = messageContext.NativeMessageId;
                     var headers = messageContext.Headers;
-
                     var log = NServiceBus.Logging.LogManager.GetLogger<ServiceControlComponentRunner>();
-                    headers.TryGetValue(Headers.MessageId, out var originalMessageId);
+                    headers.TryGetValue(Headers.MessageId,
+                        out var originalMessageId);
                     log.Debug($"OnMessage for message '{id}'({originalMessageId ?? string.Empty}).");
 
                     //Do not filter out CC, SA and HB messages as they can't be stamped
-                    if (headers.TryGetValue(Headers.EnclosedMessageTypes, out var messageTypes)
+                    if (headers.TryGetValue(Headers.EnclosedMessageTypes,
+                            out var messageTypes)
                         && (messageTypes.StartsWith("ServiceControl.Contracts") || messageTypes.StartsWith("ServiceControl.EndpointPlugin")))
                     {
                         return false;

src/ServiceControl.Audit/App.config

@@ -8,7 +8,7 @@ These settings are only here so that we can debug ServiceControl while developin
     <add key="ServiceControl.Audit/ServiceControlQueueAddress" value="Particular.ServiceControl" />
     <add key="ServiceControl.Audit/HostName" value="localhost" />
     <add key="ServiceControl.Audit/DatabaseMaintenancePort" value="44445" />
-
+    <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" />
     <!--<add key="ServiceControl.Audit/TransportType" value="AmazonSQS" />-->

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,6 +95,7 @@ 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;
                 }
 
@@ -168,6 +162,7 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
                     logger.Info("Shutting down. Already stopped, skipping shut down");
                     return; //Already stopped
                 }
+
                 var stoppable = queueIngestor;
                 queueIngestor = null;
                 logger.Info("Shutting down. Infrastructure shut down commencing");
@@ -196,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;
@@ -217,11 +212,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)
                 {
@@ -261,9 +256,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;

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,26 +30,11 @@ 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(),
-                new EnrichWithTrackingIds(),
-                new ProcessingStatisticsEnricher(),
-                new DetectNewEndpointsFromAuditImportsEnricher(endpointInstanceMonitoring),
-                new DetectSuccessfulRetriesEnricher(),
-                new SagaRelationshipsEnricher()
-            }.Concat(auditEnrichers).ToArray();
+            var enrichers = new IEnrichImportedAuditMessages[] { new MessageTypeEnricher(), new EnrichWithTrackingIds(), new ProcessingStatisticsEnricher(), new DetectNewEndpointsFromAuditImportsEnricher(endpointInstanceMonitoring), new DetectSuccessfulRetriesEnricher(), new SagaRelationshipsEnricher() }.Concat(auditEnrichers).ToArray();
 
             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)
@@ -71,6 +54,7 @@ public async Task Ingest(List<MessageContext> contexts)
                     {
                         Log.Debug($"Forwarding {stored.Count} messages");
                     }
+
                     await Forward(stored, logQueueAddress);
                     if (Log.IsDebugEnabled)
                     {
@@ -159,7 +143,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>();
     }
 }

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";
+}

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,29 +16,13 @@
     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();
@@ -51,7 +36,6 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
             IAuditIngestionUnitOfWork unitOfWork = null;
             try
             {
-
                 // deliberately not using the using statement because we dispose async explicitly
                 unitOfWork = await unitOfWorkFactory.StartNew(contexts.Count);
                 var inserts = new List<Task>(contexts.Count);
@@ -89,12 +73,13 @@ 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))
                     {
@@ -103,12 +88,13 @@ public async Task<IReadOnlyList<MessageContext>> Persist(IReadOnlyList<MessageCo
                             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);
@@ -146,10 +132,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)
                     {
@@ -263,6 +249,7 @@ async Task ProcessAuditMessage(MessageContext context)
                 {
                     Logger.Debug($"Emitting {commandsToEmit.Count} commands and {messagesToEmit.Count} control messages.");
                 }
+
                 foreach (var commandToEmit in commandsToEmit)
                 {
                     await messageSession.Send(commandToEmit);
@@ -301,16 +288,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>();
     }
 }