Add more metrics

Author: andreasohlund

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -2,11 +2,11 @@
 {
     using System;
     using System.Collections.Generic;
-    using System.Diagnostics.Metrics;
     using System.Threading;
     using System.Threading.Channels;
     using System.Threading.Tasks;
     using Infrastructure.Settings;
+    using Metrics;
     using Microsoft.Extensions.Hosting;
     using NServiceBus;
     using NServiceBus.Logging;
@@ -26,7 +26,8 @@ public AuditIngestion(
             AuditIngestionCustomCheck.State ingestionState,
             AuditIngestor auditIngestor,
             IAuditIngestionUnitOfWorkFactory unitOfWorkFactory,
-            IHostApplicationLifetime applicationLifetime)
+            IHostApplicationLifetime applicationLifetime,
+            AuditIngestionMetrics metrics)
         {
             inputEndpoint = settings.AuditQueue;
             this.transportCustomization = transportCustomization;
@@ -35,13 +36,16 @@ public AuditIngestion(
             this.unitOfWorkFactory = unitOfWorkFactory;
             this.settings = settings;
             this.applicationLifetime = applicationLifetime;
+            this.metrics = metrics;
 
             if (!transportSettings.MaxConcurrency.HasValue)
             {
                 throw new ArgumentException("MaxConcurrency is not set in TransportSettings");
             }
 
-            channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(transportSettings.MaxConcurrency.Value)
+            MaxBatchSize = transportSettings.MaxConcurrency.Value;
+
+            channel = Channel.CreateBounded<MessageContext>(new BoundedChannelOptions(MaxBatchSize)
             {
                 SingleReader = true,
                 SingleWriter = false,
@@ -190,22 +194,21 @@ async Task EnsureStopped(CancellationToken cancellationToken)
 
         async Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken)
         {
-            var tags = Telemetry.GetIngestedMessageTags(messageContext.Headers, messageContext.Body);
-            using (new DurationRecorder(ingestionDuration, tags))
+            using var messageIngestionMetrics = metrics.BeginIngestion(messageContext);
+
+            if (settings.MessageFilter != null && settings.MessageFilter(messageContext))
             {
-                if (settings.MessageFilter != null && settings.MessageFilter(messageContext))
-                {
-                    return;
-                }
+                messageIngestionMetrics.Skipped();
+                return;
+            }
 
-                var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
-                messageContext.SetTaskCompletionSource(taskCompletionSource);
+            var taskCompletionSource = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
+            messageContext.SetTaskCompletionSource(taskCompletionSource);
 
-                await channel.Writer.WriteAsync(messageContext, cancellationToken);
-                await taskCompletionSource.Task;
+            await channel.Writer.WriteAsync(messageContext, cancellationToken);
+            _ = await taskCompletionSource.Task;
 
-                successfulMessagesCounter.Add(1, tags);
-            }
+            messageIngestionMetrics.Success();
         }
 
         public override async Task StartAsync(CancellationToken cancellationToken)
@@ -218,27 +221,27 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
         {
             try
             {
-                var contexts = new List<MessageContext>(transportSettings.MaxConcurrency.Value);
+                var contexts = new List<MessageContext>(MaxBatchSize);
 
                 while (await channel.Reader.WaitToReadAsync(stoppingToken))
                 {
                     // will only enter here if there is something to read.
                     try
                     {
                         // as long as there is something to read this will fetch up to MaximumConcurrency items
-                        using (var recorder = new DurationRecorder(batchDuration))
+                        using (var batchMetrics = metrics.BeginBatch(MaxBatchSize))
                         {
                             while (channel.Reader.TryRead(out var context))
                             {
                                 contexts.Add(context);
                             }
 
-                            recorder.Tags.Add("ingestion.batch_size", contexts.Count);
-
                             await auditIngestor.Ingest(contexts);
+
+                            batchMetrics.Complete(contexts.Count);
                         }
 
-                        consecutiveBatchFailuresCounter.Record(0);
+                        //metrics.ClearB .Record(0);
                     }
                     catch (Exception e)
                     {
@@ -257,7 +260,7 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                         logger.Info("Ingesting messages failed", e);
 
                         // no need to do interlocked increment since this is running sequential
-                        consecutiveBatchFailuresCounter.Record(consecutiveBatchFailures++);
+                        //consecutiveBatchFailuresCounter.Record(consecutiveBatchFailures++);
                     }
                     finally
                     {
@@ -297,9 +300,9 @@ public override async Task StopAsync(CancellationToken cancellationToken)
         }
 
         TransportInfrastructure transportInfrastructure;
-        IMessageReceiver messageReceiver;
-        long consecutiveBatchFailures = 0;
+        IMessageReceiver queueIngestor;
 
+        readonly int MaxBatchSize;
         readonly SemaphoreSlim startStopSemaphore = new(1);
         readonly string inputEndpoint;
         readonly ITransportCustomization transportCustomization;
@@ -309,12 +312,9 @@ public override async Task StopAsync(CancellationToken cancellationToken)
         readonly IAuditIngestionUnitOfWorkFactory unitOfWorkFactory;
         readonly Settings settings;
         readonly Channel<MessageContext> channel;
-        readonly Histogram<double> batchDuration = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion", "batch_duration"), unit: "ms", "Average audit message batch processing duration");
-        readonly Counter<long> successfulMessagesCounter = Telemetry.Meter.CreateCounter<long>(Telemetry.CreateInstrumentName("ingestion", "success"), description: "Successful ingested audit message count");
-        readonly Histogram<long> consecutiveBatchFailuresCounter = Telemetry.Meter.CreateHistogram<long>(Telemetry.CreateInstrumentName("ingestion", "consecutive_batch_failures"), unit: "count", description: "Consecutive audit ingestion batch failure");
-        readonly Histogram<double> ingestionDuration = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion", "duration"), unit: "ms", description: "Average incoming audit message processing duration");
         readonly Watchdog watchdog;
         readonly IHostApplicationLifetime applicationLifetime;
+        readonly AuditIngestionMetrics metrics;
 
         static readonly ILog logger = LogManager.GetLogger<AuditIngestion>();
 

src/ServiceControl.Audit/Auditing/AuditIngestionMetrics.cs

@@ -5,6 +5,7 @@
     using System.Linq;
     using System.Threading.Tasks;
     using Infrastructure.Settings;
+    using Metrics;
     using Monitoring;
     using NServiceBus;
     using NServiceBus.Logging;

src/ServiceControl.Audit/Auditing/AuditIngestor.cs

@@ -0,0 +1,49 @@
+namespace ServiceControl.Audit.Auditing.Metrics;
+
+using System.Diagnostics.Metrics;
+using NServiceBus.Transport;
+
+public class AuditIngestionMetrics
+{
+    public AuditIngestionMetrics(IMeterFactory meterFactory)
+    {
+        var meter = meterFactory.Create(MeterName, MeterVersion);
+
+        forwardedMessagesCounter = meter.CreateCounter<long>(CreateInstrumentName("forwarded"), description: "Audit ingestion forwarded message count");
+        batchDuration = meter.CreateHistogram<double>(CreateInstrumentName("batch_duration"), unit: "ms", "Average audit message batch processing duration");
+        consecutiveBatchFailureGauge = meter.CreateObservableGauge(CreateInstrumentName("consecutive_batch_failures"), () => consecutiveBatchFailures, unit: "count", description: "Consecutive audit ingestion batch failure");
+        ingestionDuration = meter.CreateHistogram<double>(CreateInstrumentName("duration"), unit: "ms", description: "Average incoming audit message processing duration");
+    }
+
+    public void IncrementMessagesForwarded(int count) => forwardedMessagesCounter.Add(count);
+
+    public MessageIngestionMetrics BeginIngestion(MessageContext messageContext) => new(messageContext, ingestionDuration);
+
+    public BatchMetrics BeginBatch(int maxBatchSize) => new(maxBatchSize, batchDuration, RecordBatchOutcome);
+
+    void RecordBatchOutcome(bool success)
+    {
+        if (success)
+        {
+            consecutiveBatchFailures = 0;
+        }
+        else
+        {
+            consecutiveBatchFailures++;
+        }
+    }
+
+    static string CreateInstrumentName(string instrumentName) => $"sc.audit.ingestion.{instrumentName}".ToLower();
+
+    long consecutiveBatchFailures;
+
+    readonly Counter<long> forwardedMessagesCounter;
+    readonly Histogram<double> batchDuration;
+#pragma warning disable IDE0052
+    readonly ObservableGauge<long> consecutiveBatchFailureGauge;
+#pragma warning restore IDE0052
+    readonly Histogram<double> ingestionDuration;
+
+    const string MeterName = "Particular.ServiceControl.Audit";
+    const string MeterVersion = "0.1.0";
+}

src/ServiceControl.Audit/Auditing/Metrics/AuditIngestionMetrics.cs

@@ -0,0 +1,35 @@
+namespace ServiceControl.Audit.Auditing.Metrics;
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+public record BatchMetrics(int MaxBatchSize, Histogram<double> BatchDuration, Action<bool> IsSuccess) : IDisposable
+{
+    public void Dispose()
+    {
+        var tags = new TagList();
+
+        string result;
+
+        if (actualBatchSize <= 0)
+        {
+            result = "failed";
+            IsSuccess(false);
+        }
+        else
+        {
+            result = actualBatchSize == MaxBatchSize ? "full" : "partial";
+
+            IsSuccess(true);
+        }
+
+        tags.Add("result", result);
+        BatchDuration.Record(sw.ElapsedMilliseconds, tags);
+    }
+
+    public void Complete(int size) => actualBatchSize = size;
+
+    int actualBatchSize = -1;
+    readonly Stopwatch sw = Stopwatch.StartNew();
+}

src/ServiceControl.Audit/Auditing/Metrics/BatchMetrics.cs

@@ -0,0 +1,45 @@
+namespace ServiceControl.Audit.Auditing.Metrics;
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using EndpointPlugin.Messages.SagaState;
+using NServiceBus;
+using NServiceBus.Transport;
+
+public record MessageIngestionMetrics(MessageContext Message, Histogram<double> Duration) : IDisposable
+{
+    public void Skipped() => result = "skipped";
+
+    public void Success() => result = "success";
+
+    public void Dispose()
+    {
+        var tags = GetTags(Message);
+
+        tags.Add("result", result);
+        Duration.Record(sw.ElapsedMilliseconds, tags);
+    }
+
+    static TagList GetTags(MessageContext messageContext)
+    {
+        var tags = new TagList();
+
+        if (messageContext.Headers.TryGetValue(Headers.EnclosedMessageTypes, out var messageType))
+        {
+            tags.Add("message.category", messageType == SagaUpdateMessageType ? "saga-update" : "audit-message");
+        }
+        else
+        {
+            tags.Add("message.category", "control-message");
+        }
+
+        return tags;
+    }
+
+    string result = "failed";
+
+    readonly Stopwatch sw = Stopwatch.StartNew();
+
+    static readonly string SagaUpdateMessageType = typeof(SagaUpdatedMessage).FullName;
+}

src/ServiceControl.Audit/Auditing/Metrics/MessageIngestionMetrics.cs

@@ -6,6 +6,7 @@ namespace ServiceControl.Audit;
 using System.Threading.Tasks;
 using Auditing;
 using Hosting;
+using Auditing.Metrics;
 using Infrastructure;
 using Infrastructure.Settings;
 using Microsoft.AspNetCore.HttpLogging;