Report error ingestion through OpenTelemetry (3/4)

src/ServiceControl.Audit.UnitTests/Auditing/Metrics/IngestionMetricsTests.cs

@@ -0,0 +1,98 @@
+namespace ServiceControl.Audit.UnitTests.Auditing.Metrics;
+
+using System.Collections.Generic;
+using System.Diagnostics.Metrics;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.DependencyInjection;
+using NUnit.Framework;
+using ServiceControl.Audit.Auditing.Metrics;
+
+/// <summary>
+/// Instrument names are what dashboards and alerts are built on, so they are a published contract
+/// and not an implementation detail.
+/// </summary>
+[TestFixture]
+class IngestionMetricsTests
+{
+    [SetUp]
+    public void CreateMeterFactory() => provider = new ServiceCollection().AddMetrics().BuildServiceProvider();
+
+    [TearDown]
+    public void DisposeMeterFactory() => provider.Dispose();
+
+    [Test]
+    public void The_meter_publishes_the_instruments_it_is_named_for()
+    {
+        var published = new List<string>();
+
+        using var listener = new MeterListener
+        {
+            InstrumentPublished = (instrument, _) =>
+            {
+                if (BelongsToThisTest(instrument))
+                {
+                    published.Add(instrument.Name);
+                }
+            }
+        };
+
+        listener.Start();
+
+        _ = new IngestionMetrics(MeterFactory);
+
+        Assert.That(published.Order(), Is.EqualTo(new[]
+        {
+            "sc.audit.ingestion.batch_duration_seconds",
+            "sc.audit.ingestion.consecutive_batch_failures_total",
+            "sc.audit.ingestion.failures_total",
+            "sc.audit.ingestion.message_duration_seconds"
+        }));
+    }
+
+    [Test]
+    public void Concurrent_batch_failures_are_all_counted()
+    {
+        var metrics = new IngestionMetrics(MeterFactory);
+
+        const int failedBatches = 1000;
+
+        Parallel.For(0, failedBatches, _ =>
+        {
+            using var batch = metrics.BeginBatch(maxBatchSize: 1);
+        });
+
+        Assert.That(ReadConsecutiveBatchFailures(), Is.EqualTo(failedBatches));
+    }
+
+    long ReadConsecutiveBatchFailures()
+    {
+        long value = -1;
+
+        using var listener = new MeterListener
+        {
+            InstrumentPublished = (instrument, activeListener) =>
+            {
+                if (BelongsToThisTest(instrument) && instrument.Name == "sc.audit.ingestion.consecutive_batch_failures_total")
+                {
+                    activeListener.EnableMeasurementEvents(instrument);
+                }
+            }
+        };
+
+        listener.SetMeasurementEventCallback<long>((_, measurement, _, _) => value = measurement);
+        listener.Start();
+        listener.RecordObservableInstruments();
+
+        return value;
+    }
+
+    // Every fixture in the run shares the meter name, so the factory is what tells the instruments
+    // created here apart from the ones another test left behind.
+    bool BelongsToThisTest(Instrument instrument) =>
+        instrument.Meter.Name == IngestionMetrics.MeterName && ReferenceEquals(instrument.Meter.Scope, MeterFactory);
+
+    IMeterFactory MeterFactory => provider.GetRequiredService<IMeterFactory>();
+
+    ServiceProvider provider;
+}

src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs

@@ -38,12 +38,12 @@ public AuditIngestionFaultPolicy(
 
     public async Task<ErrorHandleResult> OnError(ErrorContext errorContext, CancellationToken cancellationToken = default)
     {
-        using var errorMetrics = metrics.BeginErrorHandling(errorContext);
+        using var failureMetrics = metrics.BeginErrorHandling(errorContext);
 
         //Same as recoverability policy in NServiceBusFactory
         if (errorContext.ImmediateProcessingFailures < 3)
         {
-            errorMetrics.Retry();
+            failureMetrics.Retry();
             return ErrorHandleResult.RetryRequired;
         }
 

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

@@ -3,9 +3,11 @@ namespace ServiceControl.Audit.Auditing.Metrics;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.Metrics;
+using System.Threading;
 using EndpointPlugin.Messages.SagaState;
 using NServiceBus;
 using NServiceBus.Transport;
+using ServiceControl.Infrastructure.Ingestion.Metrics;
 
 public class IngestionMetrics
 {
@@ -20,17 +22,17 @@ public IngestionMetrics(IMeterFactory meterFactory)
 
         batchDuration = meter.CreateHistogram<double>(BatchDurationInstrumentName, unit: "seconds", "Message batch processing duration in seconds");
         ingestionDuration = meter.CreateHistogram<double>(MessageDurationInstrumentName, unit: "seconds", description: "Audit message processing duration in seconds");
-        consecutiveBatchFailureGauge = meter.CreateObservableGauge($"{InstrumentPrefix}.consecutive_batch_failures_total", () => consecutiveBatchFailures, description: "Consecutive audit ingestion batch failures");
+        consecutiveBatchFailureGauge = meter.CreateObservableGauge($"{InstrumentPrefix}.consecutive_batch_failures_total", () => Volatile.Read(ref consecutiveBatchFailures), description: "Consecutive audit ingestion batch failures");
         failureCounter = meter.CreateCounter<long>($"{InstrumentPrefix}.failures_total", description: "Audit ingestion failure count");
     }
 
-    public MessageMetrics BeginIngestion(MessageContext messageContext) => new(messageContext, ingestionDuration);
+    public MessageMetrics BeginIngestion(MessageContext messageContext) => new(GetMessageTags(messageContext.Headers), ingestionDuration);
 
-    public ErrorMetrics BeginErrorHandling(ErrorContext errorContext) => new(errorContext, failureCounter);
+    public FailureMetrics BeginErrorHandling(ErrorContext errorContext) => new(GetMessageTags(errorContext.Headers), failureCounter);
 
     public BatchMetrics BeginBatch(int maxBatchSize) => new(maxBatchSize, batchDuration, RecordBatchOutcome);
 
-    public static TagList GetMessageTags(Dictionary<string, string> headers)
+    static TagList GetMessageTags(Dictionary<string, string> headers)
     {
         var tags = new TagList();
 
@@ -50,11 +52,11 @@ void RecordBatchOutcome(bool success)
     {
         if (success)
         {
-            consecutiveBatchFailures = 0;
+            Interlocked.Exchange(ref consecutiveBatchFailures, 0);
         }
         else
         {
-            consecutiveBatchFailures++;
+            Interlocked.Increment(ref consecutiveBatchFailures);
         }
     }
 

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

@@ -1,19 +1,19 @@
 namespace ServiceControl.Audit.Auditing.Metrics;
 
 using OpenTelemetry.Metrics;
+using ServiceControl.Infrastructure.Ingestion.Metrics;
 
 public static class IngestionMetricsConfiguration
 {
     public static void AddIngestionMetrics(this MeterProviderBuilder builder)
     {
         builder.AddMeter(IngestionMetrics.MeterName);
 
-        // Note: Views can be replaced by new InstrumentAdvice<double> { HistogramBucketBoundaries = [...] }; once we can update to the latest OpenTelemetry packages
         builder.AddView(
             instrumentName: IngestionMetrics.MessageDurationInstrumentName,
-            new ExplicitBucketHistogramConfiguration { Boundaries = [0.01, 0.05, 0.1, 0.5, 1, 5] });
+            new ExplicitBucketHistogramConfiguration { Boundaries = IngestionDurations.BucketBoundaries });
         builder.AddView(
             instrumentName: IngestionMetrics.BatchDurationInstrumentName,
-            new ExplicitBucketHistogramConfiguration { Boundaries = [0.01, 0.05, 0.1, 0.5, 1, 5] });
+            new ExplicitBucketHistogramConfiguration { Boundaries = IngestionDurations.BucketBoundaries });
     }
 }

src/ServiceControl.Infrastructure/Ingestion/Metrics/BatchMetrics.cs

@@ -0,0 +1,30 @@
+namespace ServiceControl.Infrastructure.Ingestion.Metrics;
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+/// <summary>
+/// One batch write. Leaving the scope without calling <see cref="Complete" /> is what records the
+/// batch as failed, so nothing has to be told about the exception that ended it.
+/// </summary>
+public sealed class BatchMetrics(int maxBatchSize, Histogram<double> batchDuration, Action<bool> recordOutcome) : IDisposable
+{
+    public void Complete(int batchSize) => completedSize = batchSize;
+
+    public void Dispose()
+    {
+        var succeeded = completedSize > 0;
+
+        recordOutcome(succeeded);
+
+        var result = succeeded
+            ? completedSize == maxBatchSize ? "full" : "partial"
+            : "failed";
+
+        batchDuration.Record(stopwatch.Elapsed.TotalSeconds, new TagList { { "result", result } });
+    }
+
+    int completedSize = -1;
+    readonly Stopwatch stopwatch = Stopwatch.StartNew();
+}

src/ServiceControl.Infrastructure/Ingestion/Metrics/DurationScope.cs

@@ -0,0 +1,15 @@
+namespace ServiceControl.Infrastructure.Ingestion.Metrics;
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+/// <summary>
+/// How long the scope was open, in seconds, and nothing else.
+/// </summary>
+public sealed class DurationScope(Histogram<double> duration) : IDisposable
+{
+    public void Dispose() => duration.Record(stopwatch.Elapsed.TotalSeconds);
+
+    readonly Stopwatch stopwatch = Stopwatch.StartNew();
+}

src/ServiceControl.Infrastructure/Ingestion/Metrics/FailureMetrics.cs

@@ -0,0 +1,24 @@
+namespace ServiceControl.Infrastructure.Ingestion.Metrics;
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+/// <summary>
+/// One message the ingestion could not handle. Leaving the scope without saying otherwise records
+/// it as having been given up on and stored as a failed import.
+/// </summary>
+public sealed class FailureMetrics(TagList messageTags, Counter<long> failures) : IDisposable
+{
+    public void Retry() => retry = true;
+
+    public void Dispose()
+    {
+        var tags = messageTags;
+        tags.Add("result", retry ? "retry" : "stored-poison");
+
+        failures.Add(1, tags);
+    }
+
+    bool retry;
+}

src/ServiceControl.Infrastructure/Ingestion/Metrics/IngestionDurations.cs

@@ -0,0 +1,12 @@
+namespace ServiceControl.Infrastructure.Ingestion.Metrics;
+
+/// <summary>
+/// The histogram buckets every ingestion duration is reported in, shared so the instances stay
+/// comparable on one dashboard.
+/// </summary>
+public static class IngestionDurations
+{
+    // Views can give way to new InstrumentAdvice<double> { HistogramBucketBoundaries = ... } once we
+    // can update to the latest OpenTelemetry packages
+    public static readonly double[] BucketBoundaries = [0.01, 0.05, 0.1, 0.5, 1, 5];
+}

src/ServiceControl.Infrastructure/Ingestion/Metrics/MessageMetrics.cs

@@ -0,0 +1,27 @@
+namespace ServiceControl.Infrastructure.Ingestion.Metrics;
+
+using System;
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+
+/// <summary>
+/// One message, from being received to its batch being written. Leaving the scope without saying
+/// otherwise records it as failed.
+/// </summary>
+public sealed class MessageMetrics(TagList messageTags, Histogram<double> duration) : IDisposable
+{
+    public void Skipped() => result = "skipped";
+
+    public void Success() => result = "success";
+
+    public void Dispose()
+    {
+        var tags = messageTags;
+        tags.Add("result", result);
+
+        duration.Record(stopwatch.Elapsed.TotalSeconds, tags);
+    }
+
+    string result = "failed";
+    readonly Stopwatch stopwatch = Stopwatch.StartNew();
+}

src/ServiceControl.UnitTests/ApprovalFiles/APIApprovals.PlatformSampleSettings.approved.txt

@@ -59,6 +59,7 @@
   "Port": 8888,
   "PersisterSpecificSettings": null,
   "PrintMetrics": false,
+  "OtlpEndpointUrl": null,
   "Hostname": "localhost",
   "VirtualDirectory": "",
   "HeartbeatGracePeriod": "00:00:40",