Move a few instruments to be attributes instead

Author: andreasohlund

docs/telemetry.md

@@ -18,18 +18,28 @@ The following metrics are available:
 
 ### Ingestion
 
-- `sc.audit.ingestion.success` - Successful ingested audit message count
-- `sc.audit.ingestion.retry` - Retried audit message count
-- `sc.audit.ingestion.failed` - Failed audit message count
-- `sc.audit.ingestion.duration` - Audit message processing duration (in milliseconds)
-- `sc.audit.ingestion.message_size` - Audit message body size (in kilobytes)
-- `sc.audit.ingestion.forwarded` - Forwarded audit messages count
+#### Success or failure
+
+- `sc.audit.ingestion.success` - Successful ingested audit message count (Counter)
+- `sc.audit.ingestion.retry` - Retried audit message count (Counter)
+- `sc.audit.ingestion.failed` - Failed audit message count (Counter)
+
+The above metrics also have the following attributes attached:
+
+- `messaging.message.body.size` - The size of the message body in bytes
+- `messaging.message.type` - The logical message type of the message if present
+
+#### Details
+
+- `sc.audit.ingestion.duration` - Audit message processing duration in milliseconds (Histogram)
+- `sc.audit.ingestion.forwarded` - Count of the number of forwarded audit messages if forwarding is enabled (Counter)
 
 ### Batching
 
-- `sc.audit.ingestion.batch_duration` - Batch processing duration (in milliseconds)
-- `sc.audit.ingestion.batch_size` - Batch size (number of messages)
-- `sc.audit.ingestion.consecutive_batch_failures` - Consecutive batch failures
+- `sc.audit.ingestion.batch_duration` - Batch processing duration in milliseconds (Histogram)
+  - Attributes:
+    - `ingestion.batch_size`
+- `sc.audit.ingestion.consecutive_batch_failures` - Consecutive batch failures (Counter)
 
 ## Monitoring
 

src/ServiceControl.Audit/Auditing/AuditIngestion.cs

@@ -174,7 +174,8 @@ async Task EnsureStopped(CancellationToken cancellationToken = default)
 
         async Task OnMessage(MessageContext messageContext, CancellationToken cancellationToken)
         {
-            using (new DurationRecorder(ingestionDuration))
+            var tags = Telemetry.GetIngestedMessageTags(messageContext.Headers, messageContext.Body);
+            using (new DurationRecorder(ingestionDuration, tags))
             {
                 if (settings.MessageFilter != null && settings.MessageFilter(messageContext))
                 {
@@ -187,8 +188,7 @@ async Task OnMessage(MessageContext messageContext, CancellationToken cancellati
                 await channel.Writer.WriteAsync(messageContext, cancellationToken);
                 await taskCompletionSource.Task;
 
-                successfulMessagesCounter.Add(1, Telemetry.GetIngestedMessageTags(messageContext.Headers));
-                messageSize.Record(messageContext.Body.Length / 1024.0);
+                successfulMessagesCounter.Add(1, tags);
             }
         }
 
@@ -210,14 +210,14 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken)
                     try
                     {
                         // as long as there is something to read this will fetch up to MaximumConcurrency items
-                        using (new DurationRecorder(auditBatchDuration))
+                        using (var recorder = new DurationRecorder(batchDuration))
                         {
                             while (channel.Reader.TryRead(out var context))
                             {
                                 contexts.Add(context);
                             }
 
-                            auditBatchSize.Record(contexts.Count);
+                            recorder.Tags.Add("ingestion.batch_size", contexts.Count);
 
                             await auditIngestor.Ingest(contexts);
                         }
@@ -293,9 +293,7 @@ public override async Task StopAsync(CancellationToken cancellationToken)
         readonly IAuditIngestionUnitOfWorkFactory unitOfWorkFactory;
         readonly Settings settings;
         readonly Channel<MessageContext> channel;
-        readonly Histogram<long> auditBatchSize = Telemetry.Meter.CreateHistogram<long>(Telemetry.CreateInstrumentName("ingestion", "batch_size"), description: "Audit ingestion average batch size");
-        readonly Histogram<double> auditBatchDuration = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion", "batch_duration"), unit: "ms", "Average audit message batch processing duration");
-        readonly Histogram<double> messageSize = Telemetry.Meter.CreateHistogram<double>(Telemetry.CreateInstrumentName("ingestion", "message_size"), unit: "kilobytes", description: "Average audit message body size");
+        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");

src/ServiceControl.Audit/Auditing/AuditIngestionFaultPolicy.cs

@@ -9,11 +9,10 @@
     using System.Threading;
     using System.Threading.Tasks;
     using Infrastructure;
-    using NServiceBus;
     using NServiceBus.Logging;
     using NServiceBus.Transport;
-    using ServiceControl.Audit.Persistence;
-    using ServiceControl.Configuration;
+    using Persistence;
+    using Configuration;
     using ServiceControl.Infrastructure;
 
     class AuditIngestionFaultPolicy
@@ -36,7 +35,7 @@ public AuditIngestionFaultPolicy(IFailedAuditStorage failedAuditStorage, Logging
 
         public async Task<ErrorHandleResult> OnError(ErrorContext errorContext, CancellationToken cancellationToken = default)
         {
-            var tags = Telemetry.GetIngestedMessageTags(errorContext.Message.Headers);
+            var tags = Telemetry.GetIngestedMessageTags(errorContext.Message.Headers, errorContext.Message.Body);
 
             //Same as recoverability policy in NServiceBusFactory
             if (errorContext.ImmediateProcessingFailures < 3)

src/ServiceControl.Audit/Auditing/AuditIngestor.cs

@@ -14,6 +14,7 @@
     using Persistence.UnitOfWork;
     using Recoverability;
     using SagaAudit;
+    using ServiceControl.Infrastructure;
     using ServiceControl.Transports;
 
     public class AuditIngestor

src/ServiceControl.Audit/HostApplicationBuilderExtensions.cs

@@ -19,6 +19,7 @@ namespace ServiceControl.Audit;
 using NServiceBus.Transport;
 using Persistence;
 using Transports;
+using ServiceControl.Infrastructure;
 using OpenTelemetry.Metrics;
 using OpenTelemetry.Resources;
 

src/ServiceControl.Audit/Infrastructure/DurationRecorder.cs

@@ -1,12 +1,12 @@
-namespace ServiceControl.Audit;
+namespace ServiceControl.Infrastructure;
 
 using System;
 using System.Diagnostics;
 using System.Diagnostics.Metrics;
 
-record DurationRecorder(Histogram<double> Histogram) : IDisposable
+record DurationRecorder(Histogram<double> Histogram, TagList Tags = default) : IDisposable
 {
     readonly Stopwatch sw = Stopwatch.StartNew();
 
-    public void Dispose() => Histogram.Record(sw.ElapsedMilliseconds);
+    public void Dispose() => Histogram.Record(sw.ElapsedMilliseconds, Tags);
 }

src/ServiceControl.Audit/Infrastructure/Telemetry.cs

@@ -1,5 +1,6 @@
-namespace ServiceControl.Audit;
+namespace ServiceControl.Infrastructure;
 
+using System;
 using System.Collections.Generic;
 using System.Diagnostics;
 using System.Diagnostics.Metrics;
@@ -13,18 +14,15 @@ static class Telemetry
 
     public static string CreateInstrumentName(string instrumentNamespace, string instrumentName) => $"sc.audit.{instrumentNamespace}.{instrumentName}".ToLower();
 
-    public static void AddAuditIngestionMeters(this MeterProviderBuilder builder)
-    {
-        builder.AddMeter(MeterName);
-    }
+    public static void AddAuditIngestionMeters(this MeterProviderBuilder builder) => builder.AddMeter(MeterName);
 
-    public static TagList GetIngestedMessageTags(IDictionary<string, string> headers)
+    public static TagList GetIngestedMessageTags(IDictionary<string, string> headers, ReadOnlyMemory<byte> body)
     {
-        var tags = new TagList();
+        var tags = new TagList { { "messaging.message.body.size", body.Length } };
 
         if (headers.TryGetValue(Headers.EnclosedMessageTypes, out var messageType))
         {
-            tags.Add("nservicebus.message_type", messageType);
+            tags.Add("messaging.message.type", messageType);
         }
 
         return tags;