Add duration and failure counters

Author: andreasohlund

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -182,19 +182,22 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
 
         async Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken)
         {
-            if (settings.MessageFilter != null && settings.MessageFilter(messageContext))
+            using (new DurationRecorder(ingestionDuration))
             {
-                return;
-            }
+                if (settings.MessageFilter != null && settings.MessageFilter(messageContext))
+                {
+                    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;
 
-            ingestedMessagesCounter.Add(1);
-            messageSize.Record(messageContext.Body.Length / 1024.0);
+                ingestedMessagesCounter.Add(1);
+                messageSize.Record(messageContext.Body.Length / 1024.0);
+            }
         }
 
         async Task Loop()
@@ -261,6 +264,7 @@ async Task Loop()
         readonly Histogram<double> auditBatchDuration = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion.", "batch_duration"), unit: "ms");
         readonly Histogram<double> messageSize = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion.", "message_size"), unit: "kilobytes");
         readonly Counter<long> ingestedMessagesCounter = Telemetry.Meter.CreateCounter<long>(Telemetry.CreateInstrumentName("ingestion.", "count"));
+        readonly Histogram<double> ingestionDuration = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion.", "duration"), unit: "ms");
         readonly Watchdog watchdog;
         readonly Task ingestionWorker;
         readonly IHostApplicationLifetime applicationLifetime;

src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs

@@ -2,6 +2,7 @@
 {
     using System;
     using System.Diagnostics;
+    using System.Diagnostics.Metrics;
     using System.IO;
     using System.Runtime.InteropServices;
     using System.Runtime.Versioning;
@@ -37,10 +38,13 @@ public async Task<ErrorHandleResult> OnError(ErrorContext errorContext, Cancella
             //Same as recoverability policy in NServiceBusFactory
             if (errorContext.ImmediateProcessingFailures < 3)
             {
+                retryCounter.Add(1);
                 return ErrorHandleResult.RetryRequired;
             }
 
             await StoreFailedMessageDocument(errorContext, cancellationToken);
+
+            failedCounter.Add(1);
             return ErrorHandleResult.Handled;
         }
 
@@ -100,6 +104,9 @@ void WriteToEventLog(string message)
             EventLog.WriteEntry(EventSourceCreator.SourceName, message, EventLogEntryType.Error);
         }
 
+        readonly Counter<long> retryCounter = Telemetry.Meter.CreateCounter<long>(Telemetry.CreateInstrumentName("ingestion.", "retry"));
+        readonly Counter<long> failedCounter = Telemetry.Meter.CreateCounter<long>(Telemetry.CreateInstrumentName("ingestion.", "failed"));
+
         static readonly ILog log = LogManager.GetLogger<AuditIngestionFaultPolicy>();
     }
 }