diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_envelope_handler_succeeds.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_envelope_handler_succeeds.cs index 16c7d1da43b..faca6eed9bb 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_envelope_handler_succeeds.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Metrics/When_envelope_handler_succeeds.cs @@ -5,6 +5,7 @@ using System.Collections.Generic; using System.Linq; using System.Threading.Tasks; +using Microsoft.ApplicationInsights.Extensibility; using NServiceBus; using NServiceBus.AcceptanceTesting; using NServiceBus.AcceptanceTests.Core.OpenTelemetry; diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_ambient_trace_in_message_session.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_ambient_trace_in_message_session.cs index 51ae62b9a41..6c3f58c65e5 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_ambient_trace_in_message_session.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_ambient_trace_in_message_session.cs @@ -15,7 +15,7 @@ public async Task Should_attach_to_ambient_trace() using var externalActivitySource = new ActivitySource("external trace source"); using var _ = TestingActivityListener.SetupDiagnosticListener(externalActivitySource.Name); // need to have a registered listener for activities to be created - const string wrapperActivityTraceState = "test trace state"; + const string wrapperActivityTraceState = "tracekey=traceValue"; var context = await Scenario.Define() .WithEndpoint(b => b diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_incoming_message_has_baggage_header.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_incoming_message_has_baggage_header.cs index 29acf036898..ffcc39fd5a1 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_incoming_message_has_baggage_header.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_incoming_message_has_baggage_header.cs @@ -17,7 +17,7 @@ public async Task Should_propagate_baggage_to_activity() { var sendOptions = new SendOptions(); sendOptions.RouteToThisEndpoint(); - sendOptions.SetHeader(Headers.DiagnosticsBaggage, "key1=value1,key2=value2,key3="); + sendOptions.SetHeader(Headers.DiagnosticsBaggage, "key1=value1,key2=value2,key3=value3"); await session.Send(new SomeMessage(), sendOptions); }) ) @@ -29,7 +29,7 @@ public async Task Should_propagate_baggage_to_activity() VerifyBaggageItem("key1", "value1"); VerifyBaggageItem("key2", "value2"); - VerifyBaggageItem("key3", ""); + VerifyBaggageItem("key3", "value3"); return; void VerifyBaggageItem(string key, string expectedValue) diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_outgoing_activity_has_baggage.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_outgoing_activity_has_baggage.cs index 93ff05bcfbf..b6a2310ad25 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_outgoing_activity_has_baggage.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_outgoing_activity_has_baggage.cs @@ -33,6 +33,9 @@ public async Task Should_propagate_baggage_to_headers() ) .Run(); + // Default (backwards-compatible) propagation produces the legacy comma-separated, percent-encoded format. + // The W3C OWS format ("key3 = , key2 = value2, key1 = value1") is produced only when the + // NServiceBus.Core.OpenTelemetry.UseDistributedContextPropagator AppContext switch is enabled (default in v11). Assert.That(context.BaggageHeader, Is.EqualTo("key3=,key2=value2,key1=value1")); } diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_incoming_message.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_incoming_message.cs index 99fecbaf11c..d386a9906d2 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_incoming_message.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_incoming_message.cs @@ -80,5 +80,37 @@ public Task Handle(IncomingMessage message, IMessageHandlerContext context) } } + [Test] + public async Task Should_use_receive_address_in_span_name_when_opted_in() + { + await Scenario.Define() + .WithEndpoint(e => e + .When(s => s.SendLocal(new IncomingMessage()))) + .Run(); + + var incomingMessageActivities = NServiceBusActivityListener.CompletedActivities.GetReceiveMessageActivities(); + Assert.That(incomingMessageActivities, Has.Count.EqualTo(1)); + + var incomingActivity = incomingMessageActivities.Single(); + Assert.That(incomingActivity.DisplayName, Does.StartWith("process ")); + Assert.That(incomingActivity.DisplayName, Is.Not.EqualTo("process message")); + } + + public class ReceivingEndpointWithDestinationNaming : EndpointConfigurationBuilder + { + public ReceivingEndpointWithDestinationNaming() => + EndpointSetup(b => b.Tracing().UseMessageDestinationInSpanNames = true); + + [Handler] + public class MessageHandler(Context testContext) : IHandleMessages + { + public Task Handle(IncomingMessage message, IMessageHandlerContext context) + { + testContext.MarkAsCompleted(); + return Task.CompletedTask; + } + } + } + public class IncomingMessage : IMessage; } \ No newline at end of file diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_message_with_default_activity_sources.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_message_with_default_activity_sources.cs new file mode 100644 index 00000000000..a233aa178b1 --- /dev/null +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_processing_message_with_default_activity_sources.cs @@ -0,0 +1,48 @@ +namespace NServiceBus.AcceptanceTests.Core.OpenTelemetry.Traces; + +using System.Linq; +using System.Threading.Tasks; +using EndpointTemplates; +using NServiceBus.AcceptanceTesting; +using NUnit.Framework; + +public class When_processing_message_with_default_activity_sources : OpenTelemetryAcceptanceTest +{ + // Until v11, handler spans are emitted from the "NServiceBus.Core" ActivitySource by default + // for backwards compatibility. The dedicated "NServiceBus.Core.Handler" source is opt-in via + // the NServiceBus.Core.OpenTelemetry.UseHandlerActivitySource AppContext switch (default in v11). + [Test] + public async Task Should_emit_handler_span_from_main_source() + { + await Scenario.Define() + .WithEndpoint(b => + b.When(session => session.SendLocal(new SomeMessage())) + ) + .Run(); + + var invokedHandlerActivities = NServiceBusActivityListener.CompletedActivities.GetInvokedHandlerActivities(); + + Assert.That(invokedHandlerActivities, Has.Count.EqualTo(1)); + Assert.That(invokedHandlerActivities.Single().Source.Name, Is.EqualTo("NServiceBus.Core"), + "without the opt-in switch, handler spans must keep coming from the main source so existing OpenTelemetry configurations keep seeing them"); + } + + public class Context : ScenarioContext; + + public class ReceivingEndpoint : EndpointConfigurationBuilder + { + public ReceivingEndpoint() => EndpointSetup(); + + [Handler] + public class MessageHandler(Context testContext) : IHandleMessages + { + public Task Handle(SomeMessage message, IMessageHandlerContext context) + { + testContext.MarkAsCompleted(); + return Task.CompletedTask; + } + } + } + + public class SomeMessage : IMessage; +} diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_publishing_messages.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_publishing_messages.cs index 8c402afb504..4281ddc170a 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_publishing_messages.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_publishing_messages.cs @@ -179,5 +179,68 @@ public Task Handle(ThisIsAnEvent @event, IMessageHandlerContext context) } } + [Test] + public async Task Should_use_event_type_in_span_name_when_opted_in() + { + await Scenario.Define() + .WithEndpoint(b => b + .When(ctx => ctx.SomeEventSubscribed, s => s.Publish())) + .WithEndpoint(b => b.When((session, ctx) => + { + if (ctx.HasNativePubSubSupport) + { + ctx.SomeEventSubscribed = true; + } + + return Task.CompletedTask; + })) + .Run(); + + var outgoingEventActivities = NServiceBusActivityListener.CompletedActivities.GetPublishEventActivities(); + Assert.That(outgoingEventActivities, Has.Count.EqualTo(1)); + + var publishedMessage = outgoingEventActivities.Single(); + Assert.That(publishedMessage.DisplayName, Is.EqualTo("publish ThisIsAnEvent")); + } + + public class PublisherWithDestinationNaming : EndpointConfigurationBuilder + { + public PublisherWithDestinationNaming() => + EndpointSetup(b => + { + b.Tracing().UseMessageDestinationInSpanNames = true; + b.OnEndpointSubscribed((s, context) => + { + if (s.SubscriberEndpoint.Contains(Conventions.EndpointNamingConvention(typeof(SubscriberForPublisherWithDestinationNaming)))) + { + if (s.MessageType == typeof(ThisIsAnEvent).AssemblyQualifiedName) + { + context.SomeEventSubscribed = true; + } + } + }); + }); + } + + public class SubscriberForPublisherWithDestinationNaming : EndpointConfigurationBuilder + { + public SubscriberForPublisherWithDestinationNaming() => + EndpointSetup(c => { }, + metadata => + { + metadata.RegisterPublisherFor(); + }); + + [Handler] + public class ThisHandlesSomethingHandler(Context testContext) : IHandleMessages + { + public Task Handle(ThisIsAnEvent @event, IMessageHandlerContext context) + { + testContext.MarkAsCompleted(); + return Task.CompletedTask; + } + } + } + public class ThisIsAnEvent : IEvent; } \ No newline at end of file diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_messages.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_messages.cs index af0751530d7..e14df1d05e6 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_messages.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_messages.cs @@ -131,5 +131,37 @@ public Task Handle(OutgoingMessage message, IMessageHandlerContext context) } } + [Test] + public async Task Should_use_destination_in_send_span_name_when_opted_in() + { + await Scenario.Define() + .WithEndpoint(b => b + .When(s => s.SendLocal(new OutgoingMessage()))) + .Run(); + + var outgoingMessageActivities = NServiceBusActivityListener.CompletedActivities.GetSendMessageActivities(); + Assert.That(outgoingMessageActivities, Has.Count.EqualTo(1)); + + var sentMessage = outgoingMessageActivities.Single(); + Assert.That(sentMessage.DisplayName, Does.StartWith("send ")); + Assert.That(sentMessage.DisplayName, Is.Not.EqualTo("send message")); + } + + public class TestEndpointWithDestinationNaming : EndpointConfigurationBuilder + { + public TestEndpointWithDestinationNaming() => + EndpointSetup(b => b.Tracing().UseMessageDestinationInSpanNames = true); + + [Handler] + public class MessageHandler(Context testContext) : IHandleMessages + { + public Task Handle(OutgoingMessage message, IMessageHandlerContext context) + { + testContext.MarkAsCompleted(); + return Task.CompletedTask; + } + } + } + public class OutgoingMessage : IMessage; } \ No newline at end of file diff --git a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_replies.cs b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_replies.cs index e6299389c51..c6c505d9236 100644 --- a/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_replies.cs +++ b/src/NServiceBus.AcceptanceTests/Core/OpenTelemetry/Traces/When_sending_replies.cs @@ -55,6 +55,40 @@ public Task Handle(OutgoingReply message, IMessageHandlerContext context) } } + [Test] + public async Task Should_use_destination_in_reply_span_name_when_opted_in() + { + await Scenario.Define() + .WithEndpoint(b => b + .When(s => s.SendLocal(new IncomingMessage()))) + .Run(); + + var outgoingMessageActivities = NServiceBusActivityListener.CompletedActivities.GetSendMessageActivities(); + Assert.That(outgoingMessageActivities, Has.Count.EqualTo(2), "2 messages are being sent"); + var replyMessage = outgoingMessageActivities[1]; + + Assert.That(replyMessage.DisplayName, Does.StartWith("reply ")); + } + + public class TestEndpointWithDestinationNaming : EndpointConfigurationBuilder + { + public TestEndpointWithDestinationNaming() => + EndpointSetup(b => b.Tracing().UseMessageDestinationInSpanNames = true); + + [Handler] + public class MessageHandler(Context testContext) : IHandleMessages, + IHandleMessages + { + public Task Handle(IncomingMessage message, IMessageHandlerContext context) => context.Reply(new OutgoingReply()); + + public Task Handle(OutgoingReply message, IMessageHandlerContext context) + { + testContext.MarkAsCompleted(); + return Task.CompletedTask; + } + } + } + public class IncomingMessage : IMessage; public class OutgoingReply : IMessage; diff --git a/src/NServiceBus.Core.Tests/ApprovalFiles/APIApprovals.ApproveNServiceBus.approved.txt b/src/NServiceBus.Core.Tests/ApprovalFiles/APIApprovals.ApproveNServiceBus.approved.txt index acbacb5d5ae..a5bdc1d3893 100644 --- a/src/NServiceBus.Core.Tests/ApprovalFiles/APIApprovals.ApproveNServiceBus.approved.txt +++ b/src/NServiceBus.Core.Tests/ApprovalFiles/APIApprovals.ApproveNServiceBus.approved.txt @@ -632,6 +632,11 @@ namespace NServiceBus StopApplication = 0, Continue = 1, } + public class InstrumentationOptions + { + public InstrumentationOptions() { } + public bool UseMessageDestinationInSpanNames { get; set; } + } public sealed class KeyedServiceKey { public const string Any = "______________"; @@ -772,6 +777,7 @@ namespace NServiceBus { public static void ContinueExistingTraceOnReceive(this NServiceBus.PublishOptions publishOptions) { } public static void StartNewTraceOnReceive(this NServiceBus.SendOptions sendOptions) { } + public static NServiceBus.InstrumentationOptions Tracing(this NServiceBus.EndpointConfiguration config) { } } public static class OutboxConfigExtensions { diff --git a/src/NServiceBus.Core.Tests/OpenTelemetry/ActivityFactoryTests.cs b/src/NServiceBus.Core.Tests/OpenTelemetry/ActivityFactoryTests.cs index 6c22b89bb7b..500740cc6ba 100644 --- a/src/NServiceBus.Core.Tests/OpenTelemetry/ActivityFactoryTests.cs +++ b/src/NServiceBus.Core.Tests/OpenTelemetry/ActivityFactoryTests.cs @@ -17,7 +17,7 @@ namespace NServiceBus.Core.Tests.OpenTelemetry; [TestFixture] public class ActivityFactoryTests { - readonly ActivityFactory activityFactory = new(); + readonly ActivityFactory activityFactory = new(new InstrumentationOptions()); TestingActivityListener nsbActivityListener; @@ -29,7 +29,7 @@ public class ActivityFactoryTests class NoDiagnosticListeners { - readonly ActivityFactory activityFactory = new(); + readonly ActivityFactory activityFactory = new(new InstrumentationOptions()); [Test] public void Should_return_null_incoming_activity_when_no_listeners() diff --git a/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationCompatibilityTests.cs b/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationCompatibilityTests.cs new file mode 100644 index 00000000000..7f4332e0cde --- /dev/null +++ b/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationCompatibilityTests.cs @@ -0,0 +1,116 @@ +namespace NServiceBus.Core.Tests.OpenTelemetry; + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using Extensibility; +using NUnit.Framework; + +[TestFixture] +public class ContextPropagationCompatibilityTests +{ + [SetUp] + public void EnableDistributedContextPropagator() + { + AppContext.SetSwitch(LegacyContextPropagation.UseDistributedContextPropagatorSwitchName, true); + LegacyContextPropagation.ResetUseDistributedContextPropagator(); + } + + [TearDown] + public void ResetDistributedContextPropagator() + { + AppContext.SetSwitch(LegacyContextPropagation.UseDistributedContextPropagatorSwitchName, false); + LegacyContextPropagation.ResetUseDistributedContextPropagator(); + } + + delegate void Writer(Activity activity, Dictionary headers, ContextBag context); + delegate void Reader(Activity activity, IDictionary headers); + + static readonly Writer LegacyWrite = LegacyContextPropagation.PropagateContextToHeaders; + static readonly Reader LegacyRead = LegacyContextPropagation.PropagateContextFromHeaders; + static readonly Writer NewWrite = ContextPropagation.PropagateContextToHeaders; + static readonly Reader NewRead = ContextPropagation.PropagateContextFromHeaders; + + // A value exercising every class of special character: structural baggage delimiters + // (',' ';' '='), the escape char '%', quotes, brackets, slashes, ampersand, Unicode and + // an emoji, plus interior spaces. Deliberately has NO leading/trailing whitespace, so this + // value isolates "what happens to special characters" from the separate edge-whitespace + // issue covered by New_propagation_loses_leading_whitespace_in_a_value. + // This already includes property-like syntax (the ';' and '=' delimiters), so a value such as + // "zone=eu;sensitive" is just a subset and needs no separate case here. + const string AllSpecialCharacters = "a b,c;d=e&f'g\"h\\i(j)k{l}m[n]o%p/q?r:s@t~u|vx é ü 😀 z"; + + static Dictionary Send(string value, Writer write) + { + using var sender = new Activity(ActivityNames.OutgoingMessageActivityName); + sender.SetIdFormat(ActivityIdFormat.W3C); + sender.Start(); + sender.AddBaggage("key", value); + + var headers = new Dictionary(); + write(sender, headers, new ContextBag()); + sender.Stop(); + return headers; + } + + static string Receive(Dictionary headers, Reader read) + { + using var receiver = new Activity(ActivityNames.IncomingMessageActivityName); + receiver.SetIdFormat(ActivityIdFormat.W3C); + receiver.Start(); + read(receiver, headers); + return receiver.GetBaggageItem("key"); + } + + static string Transmit(string value, Writer write, Reader read) => Receive(Send(value, write), read); + + [Test] + public void Legacy_sender_to_new_receiver_preserves_the_value() + { + var received = Transmit(AllSpecialCharacters, LegacyWrite, NewRead); + Assert.That(received, Is.EqualTo(AllSpecialCharacters)); + } + + [Test] + public void New_sender_to_legacy_receiver_prepends_a_leading_space_but_keeps_the_special_characters() + { + var received = Transmit(AllSpecialCharacters, NewWrite, LegacyRead); + + Assert.That(received, Is.EqualTo(" " + AllSpecialCharacters), + "ignoring the leading space, every special character round-trips correctly"); + } + + [Test] + public void New_propagation_loses_leading_whitespace_in_a_value() + { + const string valueWithLeadingSpace = " hasLeadingSpace"; + + var legacyRoundTrip = Transmit(valueWithLeadingSpace, LegacyWrite, LegacyRead); + var newRoundTrip = Transmit(valueWithLeadingSpace, NewWrite, NewRead); + + using (Assert.EnterMultipleScope()) + { + Assert.That(legacyRoundTrip, Is.EqualTo(valueWithLeadingSpace), + "legacy propagation preserves leading whitespace via percent-encoding"); + Assert.That(newRoundTrip, Is.EqualTo("hasLeadingSpace"), + "new propagation strips the leading whitespace from the value"); + } + } + + [TestCase(null, "", "")] + [TestCase("", "", "")] + [TestCase(" ", "", " ")] + [TestCase(" x ", "x", " x ")] + [TestCase(" x x ", "x x", " x x ")] + public void ValidateThatLegacyPropagatorPreservesLeadingAndTrailingWhitespaceInBaggageValues(string input, string expectedNew, string expectedLegacy) + { + var outputNew = Transmit(input, NewWrite, NewRead); + var outputLegacy = Transmit(input, LegacyWrite, LegacyRead); + + using (Assert.EnterMultipleScope()) + { + Assert.That(expectedNew, Is.EqualTo(outputNew), "Native propagator isn't trimming all leading and trailing whitespaces"); + Assert.That(expectedLegacy, Is.EqualTo(outputLegacy), "Legacy propagator isn't preserving leading and trailing whitespace for backwards compatibility"); + } + } +} \ No newline at end of file diff --git a/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationDefaultBehaviorTests.cs b/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationDefaultBehaviorTests.cs new file mode 100644 index 00000000000..fbe65df4b51 --- /dev/null +++ b/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationDefaultBehaviorTests.cs @@ -0,0 +1,72 @@ +namespace NServiceBus.Core.Tests.OpenTelemetry; + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using Extensibility; +using NUnit.Framework; + +[TestFixture] +public class ContextPropagationDefaultBehaviorTests +{ + // Without the opt-in switch, the endpoint default must remain the backwards-compatible + // legacy propagator (percent-encoded, comma-separated, whitespace preserved). + [SetUp] + public void EnsureDefault() + { + AppContext.SetSwitch(LegacyContextPropagation.UseDistributedContextPropagatorSwitchName, false); + LegacyContextPropagation.ResetUseDistributedContextPropagator(); + } + + [Test] + public void Default_uses_legacy_percent_encoded_baggage_format() + { + using var activity = new Activity(ActivityNames.OutgoingMessageActivityName); + activity.SetIdFormat(ActivityIdFormat.W3C); + activity.Start(); + activity.AddBaggage("serverNode", "DF 28"); + + var headers = new Dictionary(); + ContextPropagation.PropagateContextToHeaders(activity, headers, new ContextBag()); + + Assert.That(headers[Headers.DiagnosticsBaggage], Is.EqualTo("serverNode=DF%2028")); + } + + [Test] + public void Default_does_not_throw_when_baggage_value_is_null() + { + // Reproduces https://github.com/Particular/NServiceBus/issues/6983 on the legacy propagator. + // A null baggage value must not make the legacy propagator call Uri.EscapeDataString(null). + // Calls LegacyContextPropagation directly so the assertion is independent of the AppContext switch. + using var activity = new Activity(ActivityNames.OutgoingMessageActivityName); + activity.SetIdFormat(ActivityIdFormat.W3C); + activity.Start(); + activity.AddBaggage("test", null); + + var headers = new Dictionary(); + + Assert.DoesNotThrow(() => LegacyContextPropagation.PropagateContextToHeaders(activity, headers, new ContextBag())); + Assert.That(headers[Headers.DiagnosticsBaggage], Is.EqualTo("test=")); + } + + [Test] + public void Default_round_trip_preserves_value_whitespace() + { + using var outgoing = new Activity(ActivityNames.OutgoingMessageActivityName); + outgoing.SetIdFormat(ActivityIdFormat.W3C); + outgoing.Start(); + outgoing.AddBaggage("key1", " leading-and-trailing "); + + var headers = new Dictionary(); + ContextPropagation.PropagateContextToHeaders(outgoing, headers, new ContextBag()); + + using var incoming = new Activity(ActivityNames.IncomingMessageActivityName); + incoming.SetIdFormat(ActivityIdFormat.W3C); + incoming.Start(); + ContextPropagation.PropagateContextFromHeaders(incoming, headers); + + // Legacy propagation preserves leading/trailing whitespace via percent-encoding; + // the DistributedContextPropagator (opt-in) would trim it. + Assert.That(incoming.GetBaggageItem("key1"), Is.EqualTo(" leading-and-trailing ")); + } +} diff --git a/src/NServiceBus.Core.Tests/OpenTelemetry/HandlerActivitySourceTests.cs b/src/NServiceBus.Core.Tests/OpenTelemetry/HandlerActivitySourceTests.cs new file mode 100644 index 00000000000..8af728a9c8a --- /dev/null +++ b/src/NServiceBus.Core.Tests/OpenTelemetry/HandlerActivitySourceTests.cs @@ -0,0 +1,95 @@ +#nullable enable + +namespace NServiceBus.Core.Tests.OpenTelemetry; + +using System; +using System.Collections.Immutable; +using System.Diagnostics; +using System.Linq; +using Helpers; +using NServiceBus.Pipeline; +using NUnit.Framework; + +[TestFixture] +public class HandlerActivitySourceTests +{ + readonly ActivityFactory activityFactory = new(new InstrumentationOptions()); + + TestingActivityListener mainListener; + + [SetUp] + public void SetUp() => mainListener = TestingActivityListener.SetupNServiceBusDiagnosticListener(); + + [TearDown] + public void TearDown() + { + mainListener.Dispose(); + AppContext.SetSwitch(HandlerActivitySourceSwitch.UseHandlerActivitySourceSwitchName, false); + HandlerActivitySourceSwitch.ResetUseHandlerActivitySource(); + } + + static void OptIn() + { + AppContext.SetSwitch(HandlerActivitySourceSwitch.UseHandlerActivitySourceSwitchName, true); + HandlerActivitySourceSwitch.ResetUseHandlerActivitySource(); + } + + [Test] + public void Default_emits_handler_activity_from_main_source() + { + using var ambientActivity = new Activity("ambient activity"); + ambientActivity.Start(); + + var activity = activityFactory.StartHandlerActivity(new MessageHandler { HandlerType = typeof(HandlerActivitySourceTests) }); + + Assert.That(activity, Is.Not.Null); + Assert.That(activity!.Source.Name, Is.EqualTo("NServiceBus.Core")); + } + + [Test] + public void Opt_in_emits_handler_activity_from_handler_source() + { + OptIn(); + using var handlerListener = TestingActivityListener.SetupDiagnosticListener("NServiceBus.Core.Handler"); + + using var ambientActivity = new Activity("ambient activity"); + ambientActivity.Start(); + + var activity = activityFactory.StartHandlerActivity(new MessageHandler { HandlerType = typeof(HandlerActivitySourceTests) }); + + Assert.That(activity, Is.Not.Null); + Assert.That(activity!.Source.Name, Is.EqualTo("NServiceBus.Core.Handler")); + } + + [Test] + public void Opt_in_preserves_display_name_and_handler_type_tag() + { + OptIn(); + using var handlerListener = TestingActivityListener.SetupDiagnosticListener("NServiceBus.Core.Handler"); + + using var ambientActivity = new Activity("ambient activity"); + ambientActivity.Start(); + + Type handlerType = typeof(HandlerActivitySourceTests); + var activity = activityFactory.StartHandlerActivity(new MessageHandler { HandlerType = handlerType }); + + Assert.That(activity, Is.Not.Null); + Assert.That(activity!.DisplayName, Is.EqualTo(handlerType.Name)); + var tags = activity.Tags.ToImmutableDictionary(); + Assert.That(tags[ActivityTags.HandlerType], Is.EqualTo(handlerType.FullName)); + } + + [Test] + public void Opt_in_without_handler_source_listener_does_not_create_handler_activity() + { + OptIn(); + + using var ambientActivity = new Activity("ambient activity"); + ambientActivity.Start(); + + var activity = activityFactory.StartHandlerActivity(new MessageHandler { HandlerType = typeof(HandlerActivitySourceTests) }); + + Assert.That(activity, Is.Null, "handler activity must not be created when the dedicated source has no listeners"); + Assert.That(Activity.Current, Is.SameAs(ambientActivity), "user tags must land on the parent (process message) activity"); + } +} diff --git a/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationTests.cs b/src/NServiceBus.Core.Tests/OpenTelemetry/LegacyContextPropagationTests.cs similarity index 75% rename from src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationTests.cs rename to src/NServiceBus.Core.Tests/OpenTelemetry/LegacyContextPropagationTests.cs index 6de0d1ade7a..4c62f70cd71 100644 --- a/src/NServiceBus.Core.Tests/OpenTelemetry/ContextPropagationTests.cs +++ b/src/NServiceBus.Core.Tests/OpenTelemetry/LegacyContextPropagationTests.cs @@ -9,7 +9,7 @@ using NUnit.Framework; [TestFixture] -public class ContextPropagationTests +public class LegacyContextPropagationTests { [Test] public void Propagate_activity_id_to_header() @@ -82,6 +82,22 @@ public void Overwrites_existing_propagation_header() Assert.That(activity.Id, Is.EqualTo(headers[Headers.DiagnosticsTraceParent])); } + [Test] + public void Should_not_throw_when_baggage_value_is_null() + { + // Reproduces https://github.com/Particular/NServiceBus/issues/6983 + // A baggage item with a null value used to make the hand-written propagator call + // Uri.EscapeDataString(null), throwing ArgumentNullException while sending a message. + using var activity = new Activity(ActivityNames.OutgoingMessageActivityName); + activity.SetIdFormat(ActivityIdFormat.W3C); + activity.Start(); + activity.AddBaggage("test", null); + + var headers = new Dictionary(); + + Assert.DoesNotThrow(() => ContextPropagation.PropagateContextToHeaders(activity, headers, new ContextBag())); + } + [TestCaseSource(nameof(TestCases))] public void Can_propagate_baggage_from_header_to_activity(ContextPropagationTestCase testCase) { @@ -94,7 +110,9 @@ public void Can_propagate_baggage_from_header_to_activity(ContextPropagationTest headers[Headers.DiagnosticsBaggage] = testCase.BaggageHeaderValue; } - var activity = new Activity(ActivityNames.IncomingMessageActivityName); + using var activity = new Activity(ActivityNames.IncomingMessageActivityName); + activity.SetIdFormat(ActivityIdFormat.W3C); + activity.Start(); ContextPropagation.PropagateContextFromHeaders(activity, headers); @@ -114,7 +132,9 @@ public void Can_propagate_baggage_from_activity_to_header(ContextPropagationTest var headers = new Dictionary(); - var activity = new Activity(ActivityNames.OutgoingMessageActivityName); + using var activity = new Activity(ActivityNames.OutgoingMessageActivityName); + activity.SetIdFormat(ActivityIdFormat.W3C); + activity.Start(); foreach (var baggageItem in testCase.ExpectedBaggageItems.Reverse()) { @@ -131,7 +151,7 @@ public void Can_propagate_baggage_from_activity_to_header(ContextPropagationTest { Assert.That(baggageHeaderSet, Is.True, "Should have a baggage header if there is baggage"); - Assert.That(baggageValue, Is.EqualTo(testCase.BaggageHeaderValueWithoutOptionalWhitespace), "baggage header is set but is not correct"); + Assert.That(baggageValue, Is.EqualTo(testCase.BaggageHeaderValue), "baggage header is set but is not correct"); } } else @@ -146,7 +166,9 @@ public void Can_roundtrip_baggage(ContextPropagationTestCase testCase) TestContext.Out.WriteLine($"Baggage header: {testCase.BaggageHeaderValue}"); var outgoingHeaders = new Dictionary(); - var outgoingActivity = new Activity(ActivityNames.OutgoingMessageActivityName); + using var outgoingActivity = new Activity(ActivityNames.OutgoingMessageActivityName); + outgoingActivity.SetIdFormat(ActivityIdFormat.W3C); + outgoingActivity.Start(); foreach (var baggageItem in testCase.ExpectedBaggageItems.Reverse()) { @@ -157,7 +179,9 @@ public void Can_roundtrip_baggage(ContextPropagationTestCase testCase) // Simulate wire transfer var incomingHeaders = outgoingHeaders; - var incomingActivity = new Activity(ActivityNames.IncomingMessageActivityName); + using var incomingActivity = new Activity(ActivityNames.IncomingMessageActivityName); + incomingActivity.SetIdFormat(ActivityIdFormat.W3C); + incomingActivity.Start(); ContextPropagation.PropagateContextFromHeaders(incomingActivity, incomingHeaders); @@ -176,49 +200,48 @@ public void Can_roundtrip_baggage(ContextPropagationTestCase testCase) new ContextPropagationTestCase("without any baggage"), new ContextPropagationTestCase("with a single key") - .WithBaggage("key1", "value1"), + .WithBaggage("key1", "value1") + .WithHeaderValue("key1=value1"), new ContextPropagationTestCase("with multiple keys") .WithBaggage("key1", "value1") - .WithBaggage("key2", "value2"), - - new ContextPropagationTestCase("with whitespace") - .WithBaggage("key1 ", " value1") - .WithBaggage(" key2", "value2 ") - .WithBaggage(" key3 ", " value3 "), + .WithBaggage("key2", "value2") + .WithHeaderValue("key1=value1,key2=value2"), new ContextPropagationTestCase("with properties that do not have keys") - .WithBaggage("key1", "value1;property1;property2"), + .WithBaggage("key1", "value1;property1;property2") + .WithHeaderValue("key1=value1%3Bproperty1%3Bproperty2"), new ContextPropagationTestCase("with properties that have keys") - .WithBaggage("key3", "value3; propertyKey=propertyValue"), + .WithBaggage("key3", "value3; propertyKey=propertyValue") + .WithHeaderValue("key3=value3%3B%20propertyKey%3DpropertyValue"), new ContextPropagationTestCase("with values containing whitespace") - .WithBaggage("serverNode", "DF 28"), + .WithBaggage("serverNode", "DF 28") + .WithHeaderValue("serverNode=DF%2028"), new ContextPropagationTestCase("with values containing unicode") .WithBaggage("userId", "Amélie") + .WithHeaderValue("userId=Am%C3%A9lie") }; - public class ContextPropagationTestCase + public class ContextPropagationTestCase(string caseName) { - string caseName; - Dictionary baggageItems = []; + readonly Dictionary baggageItems = []; - public ContextPropagationTestCase(string caseName) + public ContextPropagationTestCase WithBaggage(string key, string value) { - this.caseName = caseName; + baggageItems.Add(key, value); + return this; } - public ContextPropagationTestCase WithBaggage(string key, string value) + public ContextPropagationTestCase WithHeaderValue(string headerValue) { - baggageItems.Add(key, value); + BaggageHeaderValue = headerValue; return this; } - public string BaggageHeaderValue => string.Join(",", from kvp in baggageItems select $"{kvp.Key}={Uri.EscapeDataString(kvp.Value)}"); - public string BaggageHeaderValueWithoutOptionalWhitespace - => string.Join(",", from kvp in baggageItems select $"{kvp.Key.Trim()}={Uri.EscapeDataString(kvp.Value)}"); + public string BaggageHeaderValue { get; private set; } public IEnumerable> ExpectedBaggageItems => from kvp in baggageItems select new KeyValuePair( kvp.Key.Trim(), diff --git a/src/NServiceBus.Core.Tests/Pipeline/MainPipelineExecutorTests.cs b/src/NServiceBus.Core.Tests/Pipeline/MainPipelineExecutorTests.cs index f6223a25eac..ee6b76d9e3b 100644 --- a/src/NServiceBus.Core.Tests/Pipeline/MainPipelineExecutorTests.cs +++ b/src/NServiceBus.Core.Tests/Pipeline/MainPipelineExecutorTests.cs @@ -130,7 +130,7 @@ static MainPipelineExecutor CreateMainPipelineExecutor(ServiceProvider servicePr new TestableMessageOperations(), new Notification(), receivePipeline, - new ActivityFactory(), + new ActivityFactory(new InstrumentationOptions()), incomingPipelineMetrics, new EnvelopeUnwrapper([], incomingPipelineMetrics)); diff --git a/src/NServiceBus.Core.Tests/Pipeline/TestableMessageOperations.cs b/src/NServiceBus.Core.Tests/Pipeline/TestableMessageOperations.cs index ddb58d3a66e..909b9fb796a 100644 --- a/src/NServiceBus.Core.Tests/Pipeline/TestableMessageOperations.cs +++ b/src/NServiceBus.Core.Tests/Pipeline/TestableMessageOperations.cs @@ -13,7 +13,7 @@ class TestableMessageOperations : MessageOperations public Pipeline SubscribePipeline => (Pipeline)subscribePipeline; public Pipeline UnsubscribePipeline => (Pipeline)unsubscribePipeline; - public TestableMessageOperations() : base(new MessageMapper(), new Pipeline(), new Pipeline(), new Pipeline(), new Pipeline(), new Pipeline(), new ActivityFactory()) + public TestableMessageOperations() : base(new MessageMapper(), new Pipeline(), new Pipeline(), new Pipeline(), new Pipeline(), new Pipeline(), new ActivityFactory(new InstrumentationOptions())) { } diff --git a/src/NServiceBus.Core.Tests/Routing/RoutingToDispatchConnectorTests.cs b/src/NServiceBus.Core.Tests/Routing/RoutingToDispatchConnectorTests.cs index ad4d34e8037..da97e86e33f 100644 --- a/src/NServiceBus.Core.Tests/Routing/RoutingToDispatchConnectorTests.cs +++ b/src/NServiceBus.Core.Tests/Routing/RoutingToDispatchConnectorTests.cs @@ -17,7 +17,7 @@ public class RoutingToDispatchConnectorTests [Test] public async Task Should_preserve_message_state_for_one_routing_strategy_for_allocation_reasons() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); IEnumerable operations = null; var testableRoutingContext = new TestableRoutingContext { @@ -59,7 +59,7 @@ await behavior.Invoke(testableRoutingContext, context => [Test] public async Task Should_copy_message_state_for_multiple_routing_strategies() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); List operations = null; var testableRoutingContext = new TestableRoutingContext { @@ -135,7 +135,7 @@ await behavior.Invoke(testableRoutingContext, context => [Test] public async Task Should_preserve_headers_generated_by_custom_routing_strategy() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); Dictionary headers = null; await behavior.Invoke(new TestableRoutingContext { RoutingStrategies = [new HeaderModifyingRoutingStrategy()] }, context => { @@ -153,7 +153,7 @@ public async Task Should_dispatch_immediately_if_user_requested() options.RequireImmediateDispatch(); var dispatched = false; - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var message = new OutgoingMessage("ID", [], Array.Empty()); await behavior.Invoke(new RoutingContext(message, @@ -170,7 +170,7 @@ await behavior.Invoke(new RoutingContext(message, public async Task Should_dispatch_immediately_if_not_sending_from_a_handler() { var dispatched = false; - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var message = new OutgoingMessage("ID", [], Array.Empty()); await behavior.Invoke(new RoutingContext(message, @@ -187,7 +187,7 @@ await behavior.Invoke(new RoutingContext(message, public async Task Should_not_dispatch_by_default() { var dispatched = false; - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var message = new OutgoingMessage("ID", [], Array.Empty()); await behavior.Invoke(new RoutingContext(message, @@ -203,7 +203,7 @@ await behavior.Invoke(new RoutingContext(message, [Test] public async Task Should_promote_message_headers_to_pipeline_activity() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var routingContext = new TestableRoutingContext(); routingContext.Message.Headers[Headers.ContentType] = "test content type"; // one of the headers that will be mapped to tags @@ -257,7 +257,7 @@ class MyMessage : IMessage; [Test] public async Task Should_merge_receive_properties_when_declared_by_transport() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var receiveProperties = new ReceiveProperties(new Dictionary { @@ -290,7 +290,7 @@ await behavior.Invoke(routingContext, context => [Test] public async Task Should_not_override_user_set_dispatch_property() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var receiveProperties = new ReceiveProperties(new Dictionary { @@ -324,7 +324,7 @@ await behavior.Invoke(routingContext, context => [Test] public async Task Should_preserve_user_dispatch_properties_even_with_receive_properties() { - var behavior = new RoutingToDispatchConnector(); + var behavior = new RoutingToDispatchConnector(new NoOpActivityFactory()); var receiveProperties = new ReceiveProperties(new Dictionary { diff --git a/src/NServiceBus.Core/Hosting/HostingComponent.Configuration.cs b/src/NServiceBus.Core/Hosting/HostingComponent.Configuration.cs index 4ed38527c71..6326b1a5006 100644 --- a/src/NServiceBus.Core/Hosting/HostingComponent.Configuration.cs +++ b/src/NServiceBus.Core/Hosting/HostingComponent.Configuration.cs @@ -26,7 +26,7 @@ public static Configuration PrepareConfiguration(Settings settings, List a serviceCollection, settings.ShouldRunInstallers, settings.UserRegistrations, - new ActivityFactory(), + new ActivityFactory(settings.InstrumentationOptions), persistenceConfiguration, installerComponent); diff --git a/src/NServiceBus.Core/Hosting/HostingComponent.Settings.cs b/src/NServiceBus.Core/Hosting/HostingComponent.Settings.cs index f46a6101f05..8e85864c2be 100644 --- a/src/NServiceBus.Core/Hosting/HostingComponent.Settings.cs +++ b/src/NServiceBus.Core/Hosting/HostingComponent.Settings.cs @@ -95,6 +95,8 @@ public bool WriteDiagnosticsToLog get; set; } + public InstrumentationOptions InstrumentationOptions => settings.GetOrDefault() ?? new InstrumentationOptions(); + internal void ConfigureHostLogging(object? endpointIdentifier) { EndpointIdentifier = endpointIdentifier; diff --git a/src/NServiceBus.Core/OpenTelemetry/InstrumentationOptions.cs b/src/NServiceBus.Core/OpenTelemetry/InstrumentationOptions.cs new file mode 100644 index 00000000000..f920a28b93a --- /dev/null +++ b/src/NServiceBus.Core/OpenTelemetry/InstrumentationOptions.cs @@ -0,0 +1,17 @@ +#nullable enable + +namespace NServiceBus; + +/// +/// Controls opt-in OpenTelemetry instrumentation behaviors. +/// Accessed via endpointConfiguration.Tracing(). +/// +public class InstrumentationOptions +{ + /// + /// Appends the destination to span names following the OTel messaging convention + /// {messaging.operation.name} {destination}, e.g. "process orders" or "send payments". + /// Disabled by default for backward compatibility. + /// + public bool UseMessageDestinationInSpanNames { get; set; } +} diff --git a/src/NServiceBus.Core/OpenTelemetry/OpenTelemetryExtensions.cs b/src/NServiceBus.Core/OpenTelemetry/OpenTelemetryExtensions.cs index bc263ffe423..9c3ad6e63e8 100644 --- a/src/NServiceBus.Core/OpenTelemetry/OpenTelemetryExtensions.cs +++ b/src/NServiceBus.Core/OpenTelemetry/OpenTelemetryExtensions.cs @@ -2,11 +2,24 @@ namespace NServiceBus; +using System; + /// /// Gives users control over the depth of an OpenTelemetry trace. /// public static class OpenTelemetryExtensions { + /// + /// Provides access to instrumentation options for OpenTelemetry tracing. + /// + /// The endpoint configuration. + /// The instance for this endpoint. + public static InstrumentationOptions Tracing(this EndpointConfiguration config) + { + ArgumentNullException.ThrowIfNull(config); + return config.Settings.GetOrCreate(); + } + /// /// Start a new OpenTelemetry trace conversation. /// diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityDisplayNames.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityDisplayNames.cs index ca6d81aff76..cf56402f4dd 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityDisplayNames.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityDisplayNames.cs @@ -10,4 +10,9 @@ static class ActivityDisplayNames public const string UnsubscribeEvent = "unsubscribe event"; public const string SendMessage = "send message"; public const string ReplyMessage = "reply"; + + // Operation-only prefixes used when UseMessageDestinationInSpanNames is enabled + internal const string ProcessOperation = "process"; + internal const string PublishOperation = "publish"; + internal const string SendOperation = "send"; } \ No newline at end of file diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityExtensions.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityExtensions.cs index 04e287679c0..30b0bf1066f 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityExtensions.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityExtensions.cs @@ -41,6 +41,8 @@ public static void SetErrorStatus(this Activity activity, Exception ex) activity.SetStatus(ActivityStatusCode.Error, ex.Message); activity.SetTag("otel.status_code", "ERROR"); activity.SetTag("otel.status_description", ex.Message); + + activity.AddEvent(new ActivityEvent("exception", DateTimeOffset.UtcNow, [ new KeyValuePair("exception.escaped", true), diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityFactory.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityFactory.cs index 6c0202a51d7..d5dcf4ec07a 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityFactory.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivityFactory.cs @@ -8,6 +8,13 @@ namespace NServiceBus; sealed class ActivityFactory : IActivityFactory { + public ActivityFactory(InstrumentationOptions options) + { + Options = options; + } + + public InstrumentationOptions Options { get; } + public Activity? StartIncomingPipelineActivity(MessageContext context) { // CreateActivity is a no-op if there are no listeners but we are doing a fast path check @@ -66,7 +73,9 @@ sealed class ActivityFactory : IActivityFactory ContextPropagation.PropagateContextFromHeaders(activity, context.Headers); - activity.DisplayName = ActivityDisplayNames.ProcessMessage; + activity.DisplayName = Options.UseMessageDestinationInSpanNames + ? $"{ActivityDisplayNames.ProcessOperation} {context.ReceiveAddress}" + : ActivityDisplayNames.ProcessMessage; activity.SetIdFormat(ActivityIdFormat.W3C); activity.AddTag(ActivityTags.NativeMessageId, context.NativeMessageId); @@ -102,7 +111,13 @@ sealed class ActivityFactory : IActivityFactory return null; } - var activity = ActivitySources.Main.StartActivity(ActivityNames.InvokeHandlerActivityName); + // Until v11 the dedicated handler source is opt-in; existing configurations only + // subscribe to the main source and must keep receiving handler spans from it. + var source = HandlerActivitySourceSwitch.UseHandlerActivitySource + ? ActivitySources.Handler + : ActivitySources.Main; + + var activity = source.StartActivity(ActivityNames.InvokeHandlerActivityName); if (activity is null) { diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivitySources.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivitySources.cs index 8f6e26858e1..939305dd0eb 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivitySources.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/ActivitySources.cs @@ -9,4 +9,8 @@ static class ActivitySources public static readonly ActivitySource Main = new("NServiceBus.Core", "0.1.0"); + + public static readonly ActivitySource Handler = + new("NServiceBus.Core.Handler", + "0.1.0"); } \ No newline at end of file diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/ContextPropagation.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/ContextPropagation.cs index 0f1d8868270..8df14752012 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/ContextPropagation.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/ContextPropagation.cs @@ -2,86 +2,83 @@ namespace NServiceBus; -using System; using System.Collections.Generic; using System.Diagnostics; -using System.Linq; using Extensibility; static class ContextPropagation { public static void PropagateContextToHeaders(Activity? activity, Dictionary headers, ContextBag contextBag) { - if (activity is null) + // TODO: investigate if we need to improve the switch check for better performance + // Removed in v11, see obsolete_v11.cs + if (!LegacyContextPropagation.UseDistributedContextPropagator) { + LegacyContextPropagation.PropagateContextToHeaders(activity, headers, contextBag); return; } - if (activity.Id is not null) + // The following part was intentionally not extracted to a separate class to prevent + // accidental leftovers when because that the legacy propagator will be removed in v11 + if (activity is null) { - headers[Headers.DiagnosticsTraceParent] = activity.Id; + return; } - if (activity.TraceStateString is not null) - { - headers[Headers.DiagnosticsTraceState] = activity.TraceStateString; - } + DistributedContextPropagator.Current.Inject(activity, headers, Setter); - // Check whether the startnewtrace setting was set in the context, if so, add it to the headers now the trace parent was added - if (contextBag.TryGet(Headers.StartNewTrace, out var headerContent)) - { - headers[Headers.StartNewTrace] = headerContent; - } + var traceParentExists = headers.ContainsKey(Headers.DiagnosticsTraceParent); + var startNewTraceOnReceive = contextBag.TryGet(Headers.StartNewTrace, out var startNewTrace); - var baggage = string.Join(",", activity.Baggage.Select(item => $"{item.Key}={Uri.EscapeDataString(item.Value ?? string.Empty)}")); - if (!string.IsNullOrEmpty(baggage)) + if (traceParentExists && startNewTraceOnReceive) { - headers[Headers.DiagnosticsBaggage] = baggage; + headers[Headers.StartNewTrace] = startNewTrace!; } } public static void PropagateContextFromHeaders(Activity? activity, IDictionary headers) { + // Removed in v11, see obsolete_v11.cs + if (!LegacyContextPropagation.UseDistributedContextPropagator) + { + LegacyContextPropagation.PropagateContextFromHeaders(activity, headers); + return; + } + + // The following part was intentionally not extracted to a separate class to prevent + // accidental leftovers when because that the legacy propagator will be removed in v11 if (activity is null) { return; } - if (headers.TryGetValue(Headers.DiagnosticsTraceState, out var traceState)) + DistributedContextPropagator.Current.ExtractTraceIdAndState(headers, Getter, out _, out var traceState); + + if (traceState is not null) { activity.TraceStateString = traceState; } - if (headers.TryGetValue(Headers.DiagnosticsBaggage, out var baggageValue)) + var baggage = DistributedContextPropagator.Current.ExtractBaggage(headers, Getter); + + if (baggage is null) + { + return; + } + + foreach (var baggageItem in baggage) { - var baggageSpan = baggageValue.AsSpan(); - // HINT: Iterate in reverse order because Activity baggage is LIFO - while (!baggageSpan.IsEmpty) - { - var lastComma = baggageSpan.LastIndexOf(','); - ReadOnlySpan baggageItem; - - if (lastComma >= 0) - { - baggageItem = baggageSpan[(lastComma + 1)..]; - baggageSpan = baggageSpan[..lastComma]; - } - else - { - baggageItem = baggageSpan; - baggageSpan = []; - } - - var firstEquals = baggageItem.IndexOf('='); - if (firstEquals < 0 || firstEquals >= baggageItem.Length) - { - continue; - } - - var key = baggageItem[..firstEquals].Trim(); - var value = baggageItem[(firstEquals + 1)..]; - activity.AddBaggage(key.ToString(), Uri.UnescapeDataString(value)); - } + activity.AddBaggage(baggageItem.Key, baggageItem.Value); } } + + static readonly DistributedContextPropagator.PropagatorSetterCallback Setter = static (carrier, key, value) => + ((IDictionary)carrier!)[key] = value; + + static readonly DistributedContextPropagator.PropagatorGetterCallback Getter = + static (carrier, key, out value, out values) => + { + values = null; + value = ((IReadOnlyDictionary)carrier!).GetValueOrDefault(key); + }; } \ No newline at end of file diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/IActivityFactory.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/IActivityFactory.cs index 8a47acb7fdd..daa6553a2eb 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/IActivityFactory.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/IActivityFactory.cs @@ -8,6 +8,7 @@ namespace NServiceBus; interface IActivityFactory { + InstrumentationOptions Options { get; } Activity? StartIncomingPipelineActivity(MessageContext context); Activity? StartOutgoingPipelineActivity(string activityName, string displayName, IBehaviorContext outgoingContext); Activity? StartHandlerActivity(MessageHandler messageHandler); diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/NoOpActivityFactory.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/NoOpActivityFactory.cs index a36d268a3bd..fe1f2f82c12 100644 --- a/src/NServiceBus.Core/OpenTelemetry/Tracing/NoOpActivityFactory.cs +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/NoOpActivityFactory.cs @@ -8,6 +8,8 @@ namespace NServiceBus; sealed class NoOpActivityFactory : IActivityFactory { + public InstrumentationOptions Options { get; } = new InstrumentationOptions(); + public Activity? StartIncomingPipelineActivity(MessageContext context) => null; public Activity? StartOutgoingPipelineActivity(string activityName, string displayName, IBehaviorContext outgoingContext) => null; diff --git a/src/NServiceBus.Core/OpenTelemetry/Tracing/obsolete_v11.cs b/src/NServiceBus.Core/OpenTelemetry/Tracing/obsolete_v11.cs new file mode 100644 index 00000000000..499be45157c --- /dev/null +++ b/src/NServiceBus.Core/OpenTelemetry/Tracing/obsolete_v11.cs @@ -0,0 +1,189 @@ +#nullable enable + +namespace NServiceBus; + +using System; +using System.Collections.Generic; +using System.Diagnostics; +using System.Linq; +using Extensibility; +using Particular.Obsoletes; + +// ============================================================================= +// EVERYTHING IN THIS FILE IS TEMPORARY AND WILL BE REMOVED IN v11. +// +// In v10.3, switching to System.Diagnostics.DistributedContextPropagator for +// OpenTelemetry trace-context/baggage propagation changes the baggage wire +// format (W3C OWS encoding + whitespace trimming) and is therefore breaking on +// rolling upgrades. To stay backwards compatible it is opt-in via an AppContext +// switch and the legacy propagator below remains the default. +// +// In v11 the new propagator becomes the default: delete this entire file and +// remove the two `if (!ObsoleteV11.UseDistributedContextPropagator)` delegation +// blocks in ContextPropagation.cs. +// ============================================================================= +static class LegacyContextPropagation +{ + enum SwitchState : byte + { + Unchecked = 0, + Enabled = 1, + Disabled = 2 + } + + static SwitchState cachedUseDistributedContextPropagator; + + [PreObsolete("https://github.com/Particular/NServiceBus/issues/7825", + Note = "In v11, DistributedContextPropagator-based context propagation becomes the default and this switch will be removed together with the legacy propagator in obsolete_v11.cs.", + ReplacementTypeOrMember = "ContextPropagation")] + public const string UseDistributedContextPropagatorSwitchName = "NServiceBus.Core.OpenTelemetry.UseDistributedContextPropagator"; + + [PreObsolete("https://github.com/Particular/NServiceBus/issues/7825", + Note = "In v11, DistributedContextPropagator-based context propagation becomes the default and this switch will be removed together with the legacy propagator in obsolete_v11.cs.", + ReplacementTypeOrMember = "ContextPropagation")] + public static bool UseDistributedContextPropagator + { + get + { + var state = cachedUseDistributedContextPropagator; + if (state != SwitchState.Unchecked) + { + return state == SwitchState.Enabled; + } + + state = AppContext.TryGetSwitch(UseDistributedContextPropagatorSwitchName, out var isEnabled) && isEnabled + ? SwitchState.Enabled + : SwitchState.Disabled; + cachedUseDistributedContextPropagator = state; + + return state == SwitchState.Enabled; + } + } + + internal static void ResetUseDistributedContextPropagator() => cachedUseDistributedContextPropagator = SwitchState.Unchecked; + + public static void PropagateContextToHeaders(Activity? activity, Dictionary headers, ContextBag contextBag) + { + if (activity is null) + { + return; + } + + if (activity.Id is not null) + { + headers[Headers.DiagnosticsTraceParent] = activity.Id; + } + + if (activity.TraceStateString is not null) + { + headers[Headers.DiagnosticsTraceState] = activity.TraceStateString; + } + + // Check whether the startnewtrace setting was set in the context, if so, add it to the headers now the trace parent was added + if (contextBag.TryGet(Headers.StartNewTrace, out var headerContent)) + { + headers[Headers.StartNewTrace] = headerContent; + } + + var baggage = string.Join(",", activity.Baggage.Select(item => $"{item.Key}={Uri.EscapeDataString(item.Value ?? string.Empty)}")); + if (!string.IsNullOrEmpty(baggage)) + { + headers[Headers.DiagnosticsBaggage] = baggage; + } + } + + public static void PropagateContextFromHeaders(Activity? activity, IDictionary headers) + { + if (activity is null) + { + return; + } + + if (headers.TryGetValue(Headers.DiagnosticsTraceState, out var traceState)) + { + activity.TraceStateString = traceState; + } + + if (headers.TryGetValue(Headers.DiagnosticsBaggage, out var baggageValue)) + { + var baggageSpan = baggageValue.AsSpan(); + // HINT: Iterate in reverse order because Activity baggage is LIFO + while (!baggageSpan.IsEmpty) + { + var lastComma = baggageSpan.LastIndexOf(','); + ReadOnlySpan baggageItem; + + if (lastComma >= 0) + { + baggageItem = baggageSpan[(lastComma + 1)..]; + baggageSpan = baggageSpan[..lastComma]; + } + else + { + baggageItem = baggageSpan; + baggageSpan = []; + } + + var firstEquals = baggageItem.IndexOf('='); + if (firstEquals < 0 || firstEquals >= baggageItem.Length) + { + continue; + } + + var key = baggageItem[..firstEquals].Trim(); + var value = baggageItem[(firstEquals + 1)..]; + activity.AddBaggage(key.ToString(), Uri.UnescapeDataString(value)); + } + } + } +} + +// Handler spans move from the "NServiceBus.Core" ActivitySource to the dedicated +// "NServiceBus.Core.Handler" source so they can be filtered/sampled independently +// (https://github.com/Particular/NServiceBus/issues/7284). Existing OpenTelemetry +// configurations only subscribe to "NServiceBus.Core" and would silently lose handler +// spans, so the new source is opt-in via an AppContext switch until v11. +// +// In v11 the dedicated source becomes the default: delete this class and remove the +// source selection in ActivityFactory.StartHandlerActivity so it always uses +// ActivitySources.Handler. +static class HandlerActivitySourceSwitch +{ + enum SwitchState : byte + { + Unchecked = 0, + Enabled = 1, + Disabled = 2 + } + + static SwitchState cachedUseHandlerActivitySource; + + [PreObsolete("https://github.com/Particular/NServiceBus/issues/7284", + Note = "In v11, handler spans are always emitted from the NServiceBus.Core.Handler ActivitySource and this switch will be removed.", + ReplacementTypeOrMember = "ActivitySources.Handler")] + public const string UseHandlerActivitySourceSwitchName = "NServiceBus.Core.OpenTelemetry.UseHandlerActivitySource"; + + [PreObsolete("https://github.com/Particular/NServiceBus/issues/7284", + Note = "In v11, handler spans are always emitted from the NServiceBus.Core.Handler ActivitySource and this switch will be removed.", + ReplacementTypeOrMember = "ActivitySources.Handler")] + public static bool UseHandlerActivitySource + { + get + { + var state = cachedUseHandlerActivitySource; + if (state != SwitchState.Unchecked) + { + return state == SwitchState.Enabled; + } + + state = AppContext.TryGetSwitch(UseHandlerActivitySourceSwitchName, out var isEnabled) && isEnabled + ? SwitchState.Enabled + : SwitchState.Disabled; + cachedUseHandlerActivitySource = state; + + return state == SwitchState.Enabled; + } + } + + internal static void ResetUseHandlerActivitySource() => cachedUseHandlerActivitySource = SwitchState.Unchecked; +} \ No newline at end of file diff --git a/src/NServiceBus.Core/Pipeline/Outgoing/RoutingToDispatchConnector.cs b/src/NServiceBus.Core/Pipeline/Outgoing/RoutingToDispatchConnector.cs index bc4be17db8b..55f62023b4b 100644 --- a/src/NServiceBus.Core/Pipeline/Outgoing/RoutingToDispatchConnector.cs +++ b/src/NServiceBus.Core/Pipeline/Outgoing/RoutingToDispatchConnector.cs @@ -14,6 +14,13 @@ namespace NServiceBus; class RoutingToDispatchConnector : StageConnector { + readonly IActivityFactory activityFactory; + + public RoutingToDispatchConnector(IActivityFactory activityFactory) + { + this.activityFactory = activityFactory; + } + public override Task Invoke(IRoutingContext context, Func stage) { var dispatchConsistency = DispatchConsistency.Default; @@ -60,6 +67,17 @@ public override Task Invoke(IRoutingContext context, Func 0 + && operations[0].AddressTag is UnicastAddressTag unicastTag + && outgoingMessage.Headers.TryGetValue(Headers.MessageIntent, out var intentStr) + && intentStr is "Send" or "Reply") + { + activity.DisplayName = $"{activity.DisplayName} {unicastTag.Destination}"; + } } if (dispatchConsistency == DispatchConsistency.Default && context.Extensions.TryGet(out var pendingOperations)) diff --git a/src/NServiceBus.Core/Pipeline/Outgoing/SendComponent.cs b/src/NServiceBus.Core/Pipeline/Outgoing/SendComponent.cs index c00d0880656..5899d352981 100644 --- a/src/NServiceBus.Core/Pipeline/Outgoing/SendComponent.cs +++ b/src/NServiceBus.Core/Pipeline/Outgoing/SendComponent.cs @@ -27,7 +27,7 @@ public static SendComponent Initialize(PipelineSettings pipelineSettings, Hostin pipelineSettings.Register(new OutgoingPhysicalToRoutingConnector(), "Starts the message dispatch pipeline"); - pipelineSettings.Register(new RoutingToDispatchConnector(), + pipelineSettings.Register(new RoutingToDispatchConnector(hostingConfiguration.ActivityFactory), "Decides if the current message should be batched or immediately be dispatched to the transport"); pipelineSettings.Register(new BatchToDispatchConnector(), "Passes batched messages over to the immediate dispatch part of the pipeline"); pipelineSettings.Register(b => new ImmediateDispatchTerminator(b.GetRequiredService()), "Hands the outgoing messages over to the transport for immediate delivery"); diff --git a/src/NServiceBus.Core/Unicast/MessageOperations.cs b/src/NServiceBus.Core/Unicast/MessageOperations.cs index 34c60f3cda1..7f3af927a2a 100644 --- a/src/NServiceBus.Core/Unicast/MessageOperations.cs +++ b/src/NServiceBus.Core/Unicast/MessageOperations.cs @@ -65,7 +65,11 @@ async Task Publish(IBehaviorContext context, Type messageType, object message, P MergeDispatchProperties(publishContext, options.DispatchProperties); - using var activity = activityFactory.StartOutgoingPipelineActivity(ActivityNames.OutgoingEventActivityName, ActivityDisplayNames.PublishEvent, publishContext); + var publishDisplayName = activityFactory.Options.UseMessageDestinationInSpanNames + ? $"{ActivityDisplayNames.PublishOperation} {messageType.Name}" + : ActivityDisplayNames.PublishEvent; + + using var activity = activityFactory.StartOutgoingPipelineActivity(ActivityNames.OutgoingEventActivityName, publishDisplayName, publishContext); await publishPipeline.Invoke(publishContext, activity).ConfigureAwait(false); }