Introduce critical time and handler time metrics

src/NServiceBus.AcceptanceTests/ApprovalFiles/When_endpoint_is_warmed_up.Make_sure_things_are_in_DI.approved.txt

@@ -16,6 +16,7 @@ NServiceBus.Pipeline.LogicalMessageFactory - Singleton
 NServiceBus.Settings.IReadOnlySettings - Singleton
 NServiceBus.Transport.ISubscriptionManager - Singleton
 ----------- Private registrations used by Core-----------
+NServiceBus.IncomingPipelineMetrics - Singleton
 NServiceBus.InferredMessageTypeEnricherBehavior - Transient
 NServiceBus.SubscriptionReceiverBehavior - Transient
 NServiceBus.SubscriptionRouter - Singleton

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

@@ -36,11 +36,21 @@ public TestingMetricListener(string sourceName)
             ReportedMeters.AddOrUpdate(instrument.Name, measurement, (_, val) => val + measurement);
             Tags.AddOrUpdate(instrument.Name, _ => tags, (_, _) => tags);
         });
+        meterListener.SetMeasurementEventCallback((Instrument instrument,
+            double measurement,
+            ReadOnlySpan<KeyValuePair<string, object>> t,
+            object _) =>
+        {
+            TestContext.WriteLine($"{instrument.Meter.Name}\\{instrument.Name}:{measurement}");
+            var tags = t.ToArray();
+            ReportedMeters.AddOrUpdate(instrument.Name, 1, (_, val) => val + 1);
+            Tags.AddOrUpdate(instrument.Name, _ => tags, (_, _) => tags);
+        });
         meterListener.Start();
     }
 
     public static TestingMetricListener SetupNServiceBusMetricsListener() =>
-        SetupMetricsListener("NServiceBus.Core");
+        SetupMetricsListener("NServiceBus.Core.Pipeline.Incoming");
 
     public static TestingMetricListener SetupMetricsListener(string sourceName)
     {
@@ -82,4 +92,13 @@ public object AssertTagKeyExists(string metricName, string tagKey)
 
         return meterTag.Value;
     }
+
+    public void AssertTags(string metricName, Dictionary<string, object> expectedTags)
+    {
+        foreach (var kvp in expectedTags)
+        {
+            var actualTagValue = AssertTagKeyExists(metricName, kvp.Key);
+            Assert.AreEqual(kvp.Value, actualTagValue);
+        }
+    }
 }

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

@@ -1,5 +1,6 @@
 namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics;
 
+using System.Collections.Generic;
 using System.Threading;
 using System.Threading.Tasks;
 using NServiceBus;
@@ -15,7 +16,7 @@ public async Task Should_report_successful_message_metric()
         using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
 
         _ = await Scenario.Define<Context>()
-            .WithEndpoint<EndpointWithMetrics>(b => b
+            .WithEndpoint<EndpointWithMetrics>(b => b.CustomConfig(x => x.MakeInstanceUniquelyAddressable("disc"))
                 .When(async (session, ctx) =>
                 {
                     for (var x = 0; x < 5; x++)
@@ -30,13 +31,20 @@ public async Task Should_report_successful_message_metric()
         metricsListener.AssertMetric("nservicebus.messaging.fetches", 5);
         metricsListener.AssertMetric("nservicebus.messaging.failures", 0);
 
-        var successEndpoint = metricsListener.AssertTagKeyExists("nservicebus.messaging.successes", "nservicebus.queue");
-        var successType = metricsListener.AssertTagKeyExists("nservicebus.messaging.successes", "nservicebus.message_type");
-        var fetchedEndpoint = metricsListener.AssertTagKeyExists("nservicebus.messaging.fetches", "nservicebus.queue");
+        metricsListener.AssertTags("nservicebus.messaging.fetches",
+            new Dictionary<string, object>
+            {
+                ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)),
+                ["nservicebus.discriminator"] = "disc",
+            });
 
-        Assert.AreEqual(Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)), successEndpoint);
-        Assert.AreEqual(Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)), fetchedEndpoint);
-        Assert.AreEqual(typeof(OutgoingMessage).FullName, successType);
+        metricsListener.AssertTags("nservicebus.messaging.successes",
+            new Dictionary<string, object>
+            {
+                ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)),
+                ["nservicebus.discriminator"] = "disc",
+                ["nservicebus.message_type"] = typeof(OutgoingMessage).FullName,
+            });
     }
 
     [Test]

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

@@ -1,8 +1,10 @@
 namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics;
 
+using System.Collections.Generic;
 using System.Threading.Tasks;
 using AcceptanceTesting;
 using NUnit.Framework;
+using AcceptanceTesting.Customization;
 
 public class When_message_processing_fails : OpenTelemetryAcceptanceTest
 {
@@ -13,13 +15,22 @@ public async Task Should_report_failing_message_metrics()
         _ = await Scenario.Define<Context>()
             .WithEndpoint<FailingEndpoint>(e => e
                 .DoNotFailOnErrorMessages()
+                .CustomConfig(x => x.MakeInstanceUniquelyAddressable("disc"))
                 .When(s => s.SendLocal(new FailingMessage())))
             .Done(c => c.HandlerInvoked)
             .Run();
 
         metricsListener.AssertMetric("nservicebus.messaging.fetches", 1);
         metricsListener.AssertMetric("nservicebus.messaging.failures", 1);
         metricsListener.AssertMetric("nservicebus.messaging.successes", 0);
+
+        metricsListener.AssertTags("nservicebus.messaging.failures",
+            new Dictionary<string, object>
+            {
+                ["nservicebus.queue"] = Conventions.EndpointNamingConvention(typeof(FailingEndpoint)),
+                ["nservicebus.discriminator"] = "disc",
+                ["error.type"] = typeof(SimulatedException).FullName,
+            });
     }
 
     class Context : ScenarioContext

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/When_incoming_message_handled.cs

@@ -0,0 +1,150 @@
+namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry;
+
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using AcceptanceTesting;
+using Metrics;
+using NUnit.Framework;
+using Conventions = AcceptanceTesting.Customization.Conventions;
+
+public class When_incoming_message_handled : OpenTelemetryAcceptanceTest
+{
+    static readonly string HandlerTimeMetricName = "nservicebus.messaging.handler_time";
+    static readonly string CriticalTimeMetricName = "nservicebus.messaging.critical_time";
+
+    [Test]
+    public async Task Should_record_critical_time()
+    {
+        using TestingMetricListener metricsListener = await WhenMessagesHandled(() => new MyMessage());
+        metricsListener.AssertMetric(CriticalTimeMetricName, 5);
+        AssertMandatoryTags(metricsListener, CriticalTimeMetricName, typeof(MyMessage));
+    }
+
+    [Test]
+    public async Task Should_record_success_handling_time()
+    {
+        using TestingMetricListener metricsListener = await WhenMessagesHandled(() => new MyMessage());
+        metricsListener.AssertMetric(HandlerTimeMetricName, 5);
+        AssertMandatoryTags(metricsListener, HandlerTimeMetricName, typeof(MyMessage));
+        var handlerType = metricsListener.AssertTagKeyExists(HandlerTimeMetricName, "nservicebus.message_handler_type");
+        Assert.AreEqual(typeof(MyMessageHandler).FullName, handlerType);
+        var result = metricsListener.AssertTagKeyExists(HandlerTimeMetricName, "execution.result");
+        Assert.AreEqual("success", result);
+    }
+
+    [Test]
+    public async Task Should_record_failure_handling_time()
+    {
+        using TestingMetricListener metricsListener = await WhenMessagesHandled(() => new MyExceptionalMessage());
+        metricsListener.AssertMetric(HandlerTimeMetricName, 5);
+        AssertMandatoryTags(metricsListener, HandlerTimeMetricName, typeof(MyExceptionalMessage));
+        var handlerType = metricsListener.AssertTagKeyExists(HandlerTimeMetricName, "nservicebus.message_handler_type");
+        Assert.AreEqual(typeof(MyExceptionalHandler).FullName, handlerType);
+        var error = metricsListener.AssertTagKeyExists(HandlerTimeMetricName, "error.type");
+        Assert.AreEqual(typeof(Exception).FullName, error);
+        var result = metricsListener.AssertTagKeyExists(HandlerTimeMetricName, "execution.result");
+        Assert.AreEqual("failure", result);
+    }
+
+    [Test]
+    public async Task Should_not_record_critical_time_on_failure()
+    {
+        using TestingMetricListener metricsListener = await WhenMessagesHandled(() => new MyExceptionalMessage());
+        metricsListener.AssertMetric(CriticalTimeMetricName, 0);
+    }
+
+    static async Task<TestingMetricListener> WhenMessagesHandled(Func<IMessage> messageFactory)
+    {
+        TestingMetricListener metricsListener = null;
+        try
+        {
+            metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+            _ = await Scenario.Define<Context>()
+                .WithEndpoint<EndpointWithMetrics>(b =>
+                    b.DoNotFailOnErrorMessages()
+                        .CustomConfig(c => c.MakeInstanceUniquelyAddressable("discriminator"))
+                        .When(async (session, _) =>
+                        {
+                            for (var x = 0; x < 5; x++)
+                            {
+                                try
+                                {
+                                    await session.SendLocal(messageFactory.Invoke());
+                                }
+                                catch (Exception e)
+                                {
+                                    Console.WriteLine(e);
+                                }
+                            }
+                        }))
+                .Done(c => c.TotalHandledMessages == 5)
+                .Run();
+            return metricsListener;
+        }
+        catch
+        {
+            metricsListener?.Dispose();
+            throw;
+        }
+    }
+
+
+    static void AssertMandatoryTags(
+        TestingMetricListener metricsListener,
+        string metricName,
+        Type expectedMessageType)
+    {
+        var messageType = metricsListener.AssertTagKeyExists(metricName, "nservicebus.message_type");
+        Assert.AreEqual(expectedMessageType.FullName, messageType);
+        var endpoint = metricsListener.AssertTagKeyExists(metricName, "nservicebus.queue");
+        Assert.AreEqual(Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)), endpoint);
+        var discriminator = metricsListener.AssertTagKeyExists(metricName, "nservicebus.discriminator");
+        Assert.AreEqual("discriminator", discriminator);
+    }
+
+    class Context : ScenarioContext
+    {
+        public int TotalHandledMessages;
+    }
+
+    class EndpointWithMetrics : EndpointConfigurationBuilder
+    {
+        public EndpointWithMetrics() => EndpointSetup<OpenTelemetryEnabledEndpoint>();
+    }
+
+    class MyMessageHandler : IHandleMessages<MyMessage>
+    {
+        readonly Context testContext;
+
+        public MyMessageHandler(Context testContext) => this.testContext = testContext;
+
+        public Task Handle(MyMessage message, IMessageHandlerContext context)
+        {
+            Interlocked.Increment(ref testContext.TotalHandledMessages);
+            return Task.CompletedTask;
+        }
+    }
+
+    class MyExceptionalHandler : IHandleMessages<MyExceptionalMessage>
+    {
+        readonly Context testContext;
+
+        public MyExceptionalHandler(Context testContext) => this.testContext = testContext;
+
+        public Task Handle(MyExceptionalMessage message, IMessageHandlerContext context)
+        {
+            Interlocked.Increment(ref testContext.TotalHandledMessages);
+            throw new Exception();
+        }
+    }
+
+    public class MyMessage : IMessage
+    {
+    }
+
+    public class MyExceptionalMessage : IMessage
+    {
+    }
+}

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/When_incoming_message_handled_successfully.cs

@@ -0,0 +1,66 @@
+namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry;
+
+using System.Threading;
+using System.Threading.Tasks;
+using AcceptanceTesting;
+using Metrics;
+using NUnit.Framework;
+using Conventions = AcceptanceTesting.Customization.Conventions;
+
+[NonParallelizable]
+public class WhenIncomingMessageHandledSuccessfully : NServiceBusAcceptanceTest
+{
+    [Test]
+    public async Task Should_record_handling_time()
+    {
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        _ = await Scenario.Define<Context>()
+            .WithEndpoint<EndpointWithMetrics>(b =>
+                b.When(async (session, _) =>
+                {
+                    for (var x = 0; x < 5; x++)
+                    {
+                        await session.SendLocal(new MyMessage());
+                    }
+                }))
+            .Done(c => c.TotalHandledMessages == 5)
+            .Run();
+
+        string handlingTime = "nservicebus.messaging.handler_time";
+        metricsListener.AssertMetric(handlingTime, 5);
+        var messageType = metricsListener.AssertTagKeyExists(handlingTime, "nservicebus.message_type");
+        Assert.AreEqual(typeof(MyMessage).FullName, messageType);
+        var handlerType = metricsListener.AssertTagKeyExists(handlingTime, "nservicebus.message_handler_type");
+        Assert.AreEqual(typeof(MyMessageHandler).FullName, handlerType);
+        var endpoint = metricsListener.AssertTagKeyExists(handlingTime, "nservicebus.queue");
+        Assert.AreEqual(Conventions.EndpointNamingConvention(typeof(EndpointWithMetrics)), endpoint);
+    }
+
+    class Context : ScenarioContext
+    {
+        public int TotalHandledMessages;
+    }
+
+    class EndpointWithMetrics : EndpointConfigurationBuilder
+    {
+        public EndpointWithMetrics() => EndpointSetup<OpenTelemetryEnabledEndpoint>();
+    }
+
+    class MyMessageHandler : IHandleMessages<MyMessage>
+    {
+        readonly Context testContext;
+
+        public MyMessageHandler(Context testContext) => this.testContext = testContext;
+
+        public Task Handle(MyMessage message, IMessageHandlerContext context)
+        {
+            Interlocked.Increment(ref testContext.TotalHandledMessages);
+            return Task.CompletedTask;
+        }
+    }
+
+    public class MyMessage : IMessage
+    {
+    }
+}

src/NServiceBus.AcceptanceTests/Routing/MessageDrivenSubscriptions/Sub_to_scaled_out_pubs.cs

@@ -1,7 +1,6 @@
 namespace NServiceBus.AcceptanceTests.Routing.MessageDrivenSubscriptions;
 
 using System.Collections.Concurrent;
-using System.Collections.Generic;
 using System.Threading.Tasks;
 using AcceptanceTesting;
 using AcceptanceTesting.Customization;

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

@@ -0,0 +1,23 @@
+{
+  "Note": "Changes to metrics API should result in an update to NServiceBusMeter version.",
+  "ActivitySourceVersion": "0.2.0",
+  "Tags": [
+    "error.type",
+    "execution.result",
+    "nservicebus.discriminator",
+    "nservicebus.message_handler_type",
+    "nservicebus.message_handler_types",
+    "nservicebus.message_type",
+    "nservicebus.queue"
+  ],
+  "Metrics": [
+    "nservicebus.messaging.critical_time => Histogram, Unit: s",
+    "nservicebus.messaging.failures => Counter",
+    "nservicebus.messaging.fetches => Counter",
+    "nservicebus.messaging.handler_time => Histogram, Unit: s",
+    "nservicebus.messaging.successes => Counter",
+    "nservicebus.recoverability.delayed => Counter",
+    "nservicebus.recoverability.error => Counter",
+    "nservicebus.recoverability.immediate => Counter"
+  ]
+}