Gauge meter for active handler invocations

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_handlers_are_invoked.cs

@@ -0,0 +1,73 @@
+namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics;
+
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using EndpointTemplates;
+using NServiceBus;
+using AcceptanceTesting;
+using NUnit.Framework;
+using Conventions = AcceptanceTesting.Customization.Conventions;
+
+public class When_handlers_are_invoked : OpenTelemetryAcceptanceTest
+{
+    const string ActiveHandlersMetric = "nservicebus.messaging.active_handlers";
+    const int numberOfMessages = 5;
+
+    [Test]
+    public async Task Should_report_active_handlers_gauge_that_balances_once_idle()
+    {
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        _ = await Scenario.Define<Context>()
+            .WithEndpoint<EndpointWithMetrics>(b => b.CustomConfig(c =>
+                {
+                    c.MakeInstanceUniquelyAddressable("instanceId");
+                    c.LimitMessageProcessingConcurrencyTo(10);
+                }).When(async (session, ctx) =>
+                {
+                    for (var x = 0; x < numberOfMessages; x++)
+                    {
+                        await session.SendLocal(new OutgoingMessage());
+                    }
+                }))
+            .Run();
+
+        Assert.That(metricsListener.ReportedMeters.TryGetValue(ActiveHandlersMetric, out var net), Is.True,
+            $"'{ActiveHandlersMetric}' gauge should be reported");
+        Assert.That(net, Is.EqualTo(0),
+            "increments and decrements should balance once all handlers have completed");
+
+        metricsListener.AssertTags(ActiveHandlersMetric,
+            new Dictionary<string, object>
+            {
+                ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)),
+                ["nservicebus.discriminator"] = "instanceId",
+                ["nservicebus.message_type"] = typeof(OutgoingMessage).FullName,
+                ["nservicebus.message_handler_type"] = typeof(EndpointWithMetrics.MessageHandler).FullName
+            });
+    }
+
+    public class Context : ScenarioContext
+    {
+        public int OutgoingMessagesReceived;
+    }
+
+    public class EndpointWithMetrics : EndpointConfigurationBuilder
+    {
+        public EndpointWithMetrics() => EndpointSetup<DefaultServer>();
+
+        [Handler]
+        public class MessageHandler(Context testContext) : IHandleMessages<OutgoingMessage>
+        {
+            public Task Handle(OutgoingMessage message, IMessageHandlerContext context)
+            {
+                var messagesHandled = Interlocked.Increment(ref testContext.OutgoingMessagesReceived);
+                testContext.MarkAsCompleted(messagesHandled == numberOfMessages);
+                return Task.CompletedTask;
+            }
+        }
+    }
+
+    public class OutgoingMessage : IMessage;
+}

src/NServiceBus.Core.Tests/ApprovalFiles/MeterTests.Verify_MeterAPI.approved.txt

@@ -1,7 +1,7 @@
 {
   "Note": "Changes to metrics API should result in an update to NServiceBusMeter version.",
   "MetricsSourceName": "NServiceBus.Core.Pipeline.Incoming",
-  "MetricsSourceVersion": "0.2.0",
+  "MetricsSourceVersion": "0.3.0",
   "Tags": [
     "error.type",
     "execution.result",
@@ -14,6 +14,7 @@
   ],
   "Metrics": [
     "nservicebus.envelope.unwrapped => Counter",
+    "nservicebus.messaging.active_handlers => UpDownCounter",
     "nservicebus.messaging.critical_time => Histogram, Unit: s",
     "nservicebus.messaging.failures => Counter",
     "nservicebus.messaging.fetches => Counter",

src/NServiceBus.Core/Pipeline/Incoming/IncomingPipelineMetrics.cs

@@ -21,10 +21,11 @@ class IncomingPipelineMetrics
     const string RecoverabilityDelayed = "nservicebus.recoverability.delayed";
     const string RecoverabilityError = "nservicebus.recoverability.error";
     const string EnvelopeUnwrapping = "nservicebus.envelope.unwrapped";
+    const string ActiveMessageHandlers = "nservicebus.messaging.active_handlers";
 
     public IncomingPipelineMetrics(IMeterFactory meterFactory, string queueName, string discriminator)
     {
-        var meter = meterFactory.Create("NServiceBus.Core.Pipeline.Incoming", "0.2.0");
+        var meter = meterFactory.Create("NServiceBus.Core.Pipeline.Incoming", "0.3.0");
         totalProcessedSuccessfully = meter.CreateCounter<long>(TotalProcessedSuccessfully,
             description: "Total number of messages processed successfully by the endpoint.");
         totalFetched = meter.CreateCounter<long>(TotalFetched,
@@ -45,6 +46,8 @@ public IncomingPipelineMetrics(IMeterFactory meterFactory, string queueName, str
             description: "Total number of messages sent to the error queue.");
         totalEnvelopeUnwrapping = meter.CreateCounter<long>(EnvelopeUnwrapping,
             description: "Total number of unwrapping attempts by the endpoint.");
+        activeMessageHandlers = meter.CreateUpDownCounter<long>(ActiveMessageHandlers,
+            description: "Number of message handlers currently executing.");
 
         queueNameBase = queueName;
         endpointDiscriminator = discriminator;
@@ -243,6 +246,39 @@ public void RecordSendToErrorQueue(IRecoverabilityContext recoverabilityContext)
         totalSentToErrorQueue.Add(1, meterTags);
     }
 
+    public ActiveHandlerScope TrackHandlerInvocation(IInvokeHandlerContext context)
+    {
+        if (!activeMessageHandlers.Enabled)
+        {
+            return default;
+        }
+
+        var tags = BuildActiveHandlerTags(context);
+        activeMessageHandlers.Add(1, tags);
+        return new ActiveHandlerScope(activeMessageHandlers, tags);
+    }
+
+    public readonly struct ActiveHandlerScope(UpDownCounter<long>? counter, TagList tags) : IDisposable
+    {
+        public void Dispose() => counter?.Add(-1, tags);
+    }
+
+    // The increment and the matching decrement must carry identical tags so they apply to the same
+    // time series; building them in one place keeps the two Record methods in sync.
+    // The TagList is a stack struct with 8 pre-allocated slots, so adding 4 tags does not
+    // require a heap allocation.
+    static TagList BuildActiveHandlerTags(IInvokeHandlerContext context)
+    {
+        var incomingPipelineMetricTags = context.Extensions.Get<IncomingPipelineMetricTags>();
+        TagList tags;
+        incomingPipelineMetricTags.ApplyTags(ref tags, [
+            MeterTags.QueueName,
+            MeterTags.EndpointDiscriminator]);
+        tags.Add(new KeyValuePair<string, object?>(MeterTags.MessageType, context.MessageMetadata.MessageType.FullName));
+        tags.Add(new KeyValuePair<string, object?>(MeterTags.MessageHandlerType, context.MessageHandler.HandlerType.FullName));
+        return tags;
+    }
+
     public void EnvelopeUnwrappingSucceeded(MessageContext messageContext, IEnvelopeHandler type) => RecordEnvelopeUnwrapping(messageContext, type, true, null);
     public void EnvelopeUnwrappingFailed(MessageContext messageContext, IEnvelopeHandler type, Exception? exception) => RecordEnvelopeUnwrapping(messageContext, type, false, exception);
     void RecordEnvelopeUnwrapping(MessageContext messageContext, IEnvelopeHandler type, bool succeeded, Exception? exception)
@@ -276,6 +312,7 @@ void RecordEnvelopeUnwrapping(MessageContext messageContext, IEnvelopeHandler ty
     readonly Counter<long> totalDelayedRetries;
     readonly Counter<long> totalSentToErrorQueue;
     readonly Counter<long> totalEnvelopeUnwrapping;
+    readonly UpDownCounter<long> activeMessageHandlers;
     string queueNameBase;
     string endpointDiscriminator;
 }

src/NServiceBus.Core/Pipeline/Incoming/InvokeHandlerTerminator.cs

@@ -26,6 +26,7 @@ protected override async Task Terminate(IInvokeHandlerContext context)
         // Might as well abort before invoking the handler if we're shutting down
         context.CancellationToken.ThrowIfCancellationRequested();
         var startTime = DateTimeOffset.UtcNow;
+        using var handlerScope = messagingMetricsMeters.TrackHandlerInvocation(context);
         try
         {
             await messageHandler