Use the IMeterFactory approach

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/Metrics/When_retrying_messages.cs

@@ -0,0 +1,116 @@
+namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Metrics;
+
+using System;
+using System.Threading.Tasks;
+using AcceptanceTesting;
+using NUnit.Framework;
+
+public class When_retrying_messages : OpenTelemetryAcceptanceTest
+{
+    [Test]
+    public async Task Should_increment_immediate_meter()
+    {
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        await Scenario.Define<Context>()
+            .WithEndpoint<RetryingEndpoint>(e => e
+                .CustomConfig(c => c.Recoverability().Immediate(i => i.NumberOfRetries(1)))
+                .DoNotFailOnErrorMessages()
+                .When(s => s.SendLocal(new FailingMessage())))
+            .Done(c => c.InvocationCounter == 2)
+            .Run();
+
+        metricsListener.AssertMetric("nservicebus.recoverability.immediate", 1);
+        metricsListener.AssertMetric("nservicebus.recoverability.delayed", 0);
+        metricsListener.AssertMetric("nservicebus.recoverability.error", 0);
+    }
+
+    [Test]
+    public async Task Should_increment_delayed_meter()
+    {
+        //Requires.DelayedDelivery();
+
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        await Scenario.Define<Context>()
+            .WithEndpoint<RetryingEndpoint>(e => e
+                .CustomConfig(c =>
+                {
+                    c.Recoverability().Immediate(i => i.NumberOfRetries(0));
+                    c.Recoverability().Delayed(i => i.NumberOfRetries(1).TimeIncrease(TimeSpan.FromMilliseconds(1)));
+                })
+                .DoNotFailOnErrorMessages()
+                .When(s => s.SendLocal(new FailingMessage())))
+            .Done(c => c.InvocationCounter == 2)
+            .Run();
+
+        metricsListener.AssertMetric("nservicebus.recoverability.immediate", 0);
+        metricsListener.AssertMetric("nservicebus.recoverability.delayed", 1);
+        metricsListener.AssertMetric("nservicebus.recoverability.error", 0);
+    }
+
+    [Test]
+    public async Task Should_increment_error_meter()
+    {
+        using var metricsListener = TestingMetricListener.SetupNServiceBusMetricsListener();
+
+        await Scenario.Define<Context>()
+            .WithEndpoint<RetryingEndpoint>(e => e
+                .CustomConfig(c =>
+                {
+                    c.Recoverability().Immediate(i => i.NumberOfRetries(0));
+                    c.Recoverability().Delayed(i => i.NumberOfRetries(0));
+                })
+                .DoNotFailOnErrorMessages()
+                .When(s => s.SendLocal(new FailingMessage())))
+            .Done(c => c.FailedMessages.Count == 1)
+            .Run();
+
+        metricsListener.AssertMetric("nservicebus.recoverability.immediate", 0);
+        metricsListener.AssertMetric("nservicebus.recoverability.delayed", 0);
+        metricsListener.AssertMetric("nservicebus.recoverability.error", 1);
+    }
+
+    class Context : ScenarioContext
+    {
+        public int InvocationCounter { get; set; }
+    }
+
+    class RetryingEndpoint : EndpointConfigurationBuilder
+    {
+        public RetryingEndpoint()
+        {
+            var template = new OpenTelemetryEnabledEndpoint
+            {
+                TransportConfiguration = new ConfigureEndpointAcceptanceTestingTransport(false, true)
+            };
+            EndpointSetup(template, (endpointConfiguration, descriptor) => { });
+        }
+
+        class Handler : IHandleMessages<FailingMessage>
+        {
+            Context testContext;
+
+            public Handler(Context testContext)
+            {
+                this.testContext = testContext;
+            }
+
+            public Task Handle(FailingMessage message, IMessageHandlerContext context)
+            {
+                testContext.InvocationCounter++;
+
+                if (testContext.InvocationCounter == 1)
+                {
+                    throw new SimulatedException("first attempt fails");
+                }
+
+                return Task.CompletedTask;
+            }
+        }
+    }
+
+    public class FailingMessage : IMessage
+    {
+    }
+}

src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_publishing_messages.cs

@@ -1,10 +1,11 @@
 namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Traces;
 
 using System.Collections.Immutable;
+using System.Diagnostics;
 using System.Linq;
 using System.Threading.Tasks;
-using NServiceBus.AcceptanceTesting;
-using NServiceBus.AcceptanceTesting.Customization;
+using AcceptanceTesting;
+using AcceptanceTesting.Customization;
 using NUnit.Framework;
 
 public class When_publishing_messages : OpenTelemetryAcceptanceTest
@@ -36,15 +37,89 @@ public async Task Should_create_outgoing_event_span()
         Assert.IsNull(publishedMessage.ParentId, "publishes without ambient span should start a new trace");
 
         var sentMessageTags = publishedMessage.Tags.ToImmutableDictionary();
-        sentMessageTags.VerifyTag("nservicebus.message_id", context.SentMessageId);
+        sentMessageTags.VerifyTag("nservicebus.message_id", context.PublishedMessageId);
 
         Assert.IsNotNull(context.TraceParentHeader, "tracing header should be set on the published event");
     }
 
+    [Test]
+    public async Task Should_create_child_on_receive_when_requested_via_options()
+    {
+        var context = await Scenario.Define<Context>()
+            .WithEndpoint<Publisher>(b => b
+                .When(ctx => ctx.SomeEventSubscribed, s =>
+                {
+                    var publishOptions = new PublishOptions();
+                    publishOptions.ContinueExistingTraceOnReceive();
+                    return s.Publish(new ThisIsAnEvent(), publishOptions);
+                }))
+            .WithEndpoint<Subscriber>(b => b.When((session, ctx) =>
+            {
+                if (ctx.HasNativePubSubSupport)
+                {
+                    ctx.SomeEventSubscribed = true;
+                }
+
+                return Task.CompletedTask;
+            }))
+            .Done(c => c.OutgoingEventReceived)
+            .Run();
+
+        var publishMessageActivities = NServicebusActivityListener.CompletedActivities.GetPublishEventActivities();
+        var receiveMessageActivities = NServicebusActivityListener.CompletedActivities.GetReceiveMessageActivities();
+        Assert.AreEqual(1, publishMessageActivities.Count, "1 message is published as part of this test");
+        Assert.AreEqual(1, receiveMessageActivities.Count, "1 message is received as part of this test");
+
+        var publishRequest = publishMessageActivities[0];
+        var receiveRequest = receiveMessageActivities[0];
+
+        Assert.AreEqual(publishRequest.RootId, receiveRequest.RootId, "publish and receive operations are part the same root activity");
+        Assert.IsNotNull(receiveRequest.ParentId, "incoming message does have a parent");
+
+        CollectionAssert.IsEmpty(receiveRequest.Links, "receive does not have links");
+    }
+
+    [Test]
+    public async Task Should_create_new_linked_trace_on_receive_by_default()
+    {
+        var context = await Scenario.Define<Context>()
+            .WithEndpoint<Publisher>(b => b
+                .When(ctx => ctx.SomeEventSubscribed, s =>
+                {
+                    return s.Publish(new ThisIsAnEvent());
+                }))
+            .WithEndpoint<Subscriber>(b => b.When((session, ctx) =>
+            {
+                if (ctx.HasNativePubSubSupport)
+                {
+                    ctx.SomeEventSubscribed = true;
+                }
+
+                return Task.CompletedTask;
+            }))
+            .Done(c => c.OutgoingEventReceived)
+            .Run();
+
+        var publishMessageActivities = NServicebusActivityListener.CompletedActivities.GetPublishEventActivities();
+        var receiveMessageActivities = NServicebusActivityListener.CompletedActivities.GetReceiveMessageActivities();
+        Assert.AreEqual(1, publishMessageActivities.Count, "1 message is published as part of this test");
+        Assert.AreEqual(1, receiveMessageActivities.Count, "1 message is received as part of this test");
+
+        var publishRequest = publishMessageActivities[0];
+        var receiveRequest = receiveMessageActivities[0];
+
+        Assert.AreNotEqual(publishRequest.RootId, receiveRequest.RootId, "publish and receive operations are part of different root activities");
+        Assert.IsNull(receiveRequest.ParentId, "incoming message does not have a parent, it's a root");
+
+        ActivityLink link = receiveRequest.Links.FirstOrDefault();
+        Assert.IsNotNull(link, "Receive has a link");
+        Assert.AreEqual(publishRequest.TraceId, link.Context.TraceId, "receive is linked to publish operation");
+    }
+
     public class Context : ScenarioContext
     {
         public bool OutgoingEventReceived { get; set; }
-        public string SentMessageId { get; set; }
+        public string PublishedMessageId { get; set; }
         public string TraceParentHeader { get; set; }
         public bool SomeEventSubscribed { get; set; }
     }
@@ -71,33 +146,33 @@ public class Subscriber : EndpointConfigurationBuilder
     {
         public Subscriber() =>
             EndpointSetup<OpenTelemetryEnabledEndpoint>(c =>
-                {
-                },
+            {
+            },
                 metadata =>
                 {
                     metadata.RegisterPublisherFor<ThisIsAnEvent>(typeof(Publisher));
                 });
 
         public class ThisHandlesSomethingHandler : IHandleMessages<ThisIsAnEvent>
         {
-            public ThisHandlesSomethingHandler(Context testContext)
+            public ThisHandlesSomethingHandler(Context testPublishContext)
             {
-                this.testContext = testContext;
+                this.testPublishContext = testPublishContext;
             }
 
             public Task Handle(ThisIsAnEvent @event, IMessageHandlerContext context)
             {
                 if (context.MessageHeaders.TryGetValue(Headers.DiagnosticsTraceParent, out var traceParentHeader))
                 {
-                    testContext.TraceParentHeader = traceParentHeader;
+                    testPublishContext.TraceParentHeader = traceParentHeader;
                 }
 
-                testContext.SentMessageId = context.MessageId;
-                testContext.OutgoingEventReceived = true;
+                testPublishContext.PublishedMessageId = context.MessageId;
+                testPublishContext.OutgoingEventReceived = true;
                 return Task.CompletedTask;
             }
 
-            Context testContext;
+            Context testPublishContext;
         }
     }