Allow parent/child linking in opentelemetry spans

meum · meum · commit 99ae240a7fb7 · 2025-01-28T11:05:13.000+01:00
diff --git a/projects/RabbitMQ.Client.OpenTelemetry/PublicAPI.Unshipped.txt b/projects/RabbitMQ.Client.OpenTelemetry/PublicAPI.Unshipped.txt
@@ -0,0 +1 @@
+~static OpenTelemetry.Trace.OpenTelemetryExtensions.AddRabbitMQInstrumentation(this OpenTelemetry.Trace.TracerProviderBuilder builder, System.Action<RabbitMQ.Client.RabbitMQOpenTelemetryOptions> configure) -> OpenTelemetry.Trace.TracerProviderBuilder
diff --git a/projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs b/projects/RabbitMQ.Client.OpenTelemetry/TraceProviderBuilderExtensions.cs
@@ -8,16 +8,27 @@
 
 namespace OpenTelemetry.Trace
 {
+
     public static class OpenTelemetryExtensions
     {
-        public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProviderBuilder builder)
+        public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProviderBuilder builder, Action<RabbitMQOpenTelemetryOptions> configure)
         {
+            var options = new RabbitMQOpenTelemetryOptions();
+            configure?.Invoke(options);
+            RabbitMQActivitySource.LinkType = options.LinkType;
+            RabbitMQActivitySource.UseRoutingKeyAsOperationName = options.UseRoutingKeyAsOperationName;
+
             RabbitMQActivitySource.ContextExtractor = OpenTelemetryContextExtractor;
             RabbitMQActivitySource.ContextInjector = OpenTelemetryContextInjector;
             builder.AddSource("RabbitMQ.Client.*");
             return builder;
         }
 
+        public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProviderBuilder builder)
+        {
+            return AddRabbitMQInstrumentation(builder, null);
+        }
+
         private static ActivityContext OpenTelemetryContextExtractor(IReadOnlyBasicProperties props)
         {
             // Extract the PropagationContext of the upstream parent from the message headers.
diff --git a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs
@@ -52,6 +52,7 @@ public static class RabbitMQActivitySource
             DefaultContextExtractor;
 
         public static bool UseRoutingKeyAsOperationName { get; set; } = true;
+        public static OpenTelemetryLinkType LinkType { get; set; } = OpenTelemetryLinkType.AlwaysLink;
         internal static bool PublisherHasListeners => s_publisherSource.HasListeners();
 
         internal static readonly IEnumerable<KeyValuePair<string, object?>> CreationTags = new[]
@@ -115,9 +116,16 @@ public static class RabbitMQActivitySource
             }
 
             // Extract the PropagationContext of the upstream parent from the message headers.
+            ActivityContext linkedContext = LinkType == OpenTelemetryLinkType.AlwaysLink
+                ? ContextExtractor(readOnlyBasicProperties) : default;
+            ActivityContext parentContext = LinkType == OpenTelemetryLinkType.AlwaysParentChild
+                ? ContextExtractor(readOnlyBasicProperties) : default;
+
             Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(
                 UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicGet} {routingKey}" : MessagingOperationNameBasicGet, ActivityKind.Consumer,
-                ContextExtractor(readOnlyBasicProperties));
+                linkedContext, parentContext);
+
+
             if (activity != null && activity.IsAllDataRequested)
             {
                 PopulateMessagingTags(MessagingOperationTypeReceive, MessagingOperationNameBasicGet, routingKey, exchange, deliveryTag, readOnlyBasicProperties,
@@ -128,21 +136,26 @@ public static class RabbitMQActivitySource
         }
 
         internal static Activity? Deliver(string routingKey, string exchange, ulong deliveryTag,
-            IReadOnlyBasicProperties basicProperties, int bodySize)
+            IReadOnlyBasicProperties readOnlyBasicProperties, int bodySize)
         {
             if (!s_subscriberSource.HasListeners())
             {
                 return null;
             }
 
             // Extract the PropagationContext of the upstream parent from the message headers.
+            ActivityContext linkedContext = LinkType == OpenTelemetryLinkType.AlwaysLink
+                ? ContextExtractor(readOnlyBasicProperties) : default;
+            ActivityContext parentContext = LinkType == OpenTelemetryLinkType.AlwaysParentChild
+                ? ContextExtractor(readOnlyBasicProperties) : default;
+
             Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(
                 UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicDeliver} {routingKey}" : MessagingOperationNameBasicDeliver,
-                ActivityKind.Consumer, ContextExtractor(basicProperties));
+                ActivityKind.Consumer, linkedContext, parentContext);
             if (activity != null && activity.IsAllDataRequested)
             {
                 PopulateMessagingTags(MessagingOperationTypeProcess, MessagingOperationNameBasicDeliver, routingKey, exchange,
-                    deliveryTag, basicProperties, bodySize, activity);
+                    deliveryTag, readOnlyBasicProperties, bodySize, activity);
             }
 
             return activity;
@@ -157,8 +170,13 @@ public static class RabbitMQActivitySource
         private static Activity? StartLinkedRabbitMQActivity(this ActivitySource source, string name, ActivityKind kind,
             ActivityContext linkedContext = default, ActivityContext parentContext = default)
         {
+            var links = new List<ActivityLink>();
+            if (linkedContext != default)
+            {
+                links.Add(new ActivityLink(linkedContext));
+            }
             return source.CreateActivity(name, kind, parentContext: parentContext,
-                    links: new[] { new ActivityLink(linkedContext) }, idFormat: ActivityIdFormat.W3C,
+                    links: links, idFormat: ActivityIdFormat.W3C,
                     tags: CreationTags)
                 ?.Start();
         }
diff --git a/projects/RabbitMQ.Client/Impl/RabbitMQOpenTelemetryOptions.cs b/projects/RabbitMQ.Client/Impl/RabbitMQOpenTelemetryOptions.cs
@@ -0,0 +1,14 @@
+﻿namespace RabbitMQ.Client
+{
+    public enum OpenTelemetryLinkType
+    {
+        AlwaysLink,
+        AlwaysParentChild
+    }
+
+    public class RabbitMQOpenTelemetryOptions
+    {
+        public bool UseRoutingKeyAsOperationName { get; set; }
+        public OpenTelemetryLinkType LinkType { get; set; } = OpenTelemetryLinkType.AlwaysLink;
+    }
+}
diff --git a/projects/RabbitMQ.Client/PublicAPI.Unshipped.txt b/projects/RabbitMQ.Client/PublicAPI.Unshipped.txt
@@ -0,0 +1,11 @@
+RabbitMQ.Client.OpenTelemetryLinkType
+RabbitMQ.Client.OpenTelemetryLinkType.AlwaysLink = 0 -> RabbitMQ.Client.OpenTelemetryLinkType
+RabbitMQ.Client.OpenTelemetryLinkType.AlwaysParentChild = 1 -> RabbitMQ.Client.OpenTelemetryLinkType
+RabbitMQ.Client.RabbitMQOpenTelemetryOptions
+RabbitMQ.Client.RabbitMQOpenTelemetryOptions.LinkType.get -> RabbitMQ.Client.OpenTelemetryLinkType
+RabbitMQ.Client.RabbitMQOpenTelemetryOptions.LinkType.set -> void
+RabbitMQ.Client.RabbitMQOpenTelemetryOptions.RabbitMQOpenTelemetryOptions() -> void
+RabbitMQ.Client.RabbitMQOpenTelemetryOptions.UseRoutingKeyAsOperationName.get -> bool
+RabbitMQ.Client.RabbitMQOpenTelemetryOptions.UseRoutingKeyAsOperationName.set -> void
+static RabbitMQ.Client.RabbitMQActivitySource.LinkType.get -> RabbitMQ.Client.OpenTelemetryLinkType
+static RabbitMQ.Client.RabbitMQActivitySource.LinkType.set -> void
diff --git a/projects/Test/SequentialIntegration/TestOpenTelemetry.cs b/projects/Test/SequentialIntegration/TestOpenTelemetry.cs
@@ -83,20 +83,25 @@ void AssertIntTagGreaterThanZero(Activity activity, string name)
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherAndConsumerActivityTags(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(true, false)]
+        [InlineData(false, true)]
+        [InlineData(false, false)]
+        public async Task TestPublisherAndConsumerActivityTags(bool useRoutingKeyAsOperationName, bool useParentChildLinking)
         {
             var exportedItems = new List<Activity>();
             using var tracer = Sdk.CreateTracerProviderBuilder()
-                .AddRabbitMQInstrumentation()
+                .AddRabbitMQInstrumentation(options =>
+                {
+                    options.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
+                    options.LinkType = useParentChildLinking ? OpenTelemetryLinkType.AlwaysParentChild : OpenTelemetryLinkType.AlwaysLink;
+                })
                 .AddInMemoryExporter(exportedItems)
                 .Build();
             string baggageGuid = Guid.NewGuid().ToString();
             Baggage.SetBaggage("TestItem", baggageGuid);
             Assert.Equal(baggageGuid, Baggage.GetBaggage("TestItem"));
 
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             await Task.Delay(500);
             string queueName = $"{Guid.NewGuid()}";
             QueueDeclareOk q = await _channel.QueueDeclareAsync(queueName);
@@ -132,24 +137,29 @@ public async Task TestPublisherAndConsumerActivityTags(bool useRoutingKeyAsOpera
 
             await _channel.BasicCancelAsync(consumerTag);
             await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, queueName, exportedItems, true);
+            AssertActivityData(useRoutingKeyAsOperationName, useParentChildLinking, queueName, exportedItems, true);
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherAndConsumerActivityTagsAsync(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(true, false)]
+        [InlineData(false, true)]
+        [InlineData(false, false)]
+        public async Task TestPublisherAndConsumerActivityTagsAsync(bool useRoutingKeyAsOperationName, bool useParentChildLinking)
         {
             var exportedItems = new List<Activity>();
             using var tracer = Sdk.CreateTracerProviderBuilder()
-                .AddRabbitMQInstrumentation()
+                .AddRabbitMQInstrumentation(options =>
+                {
+                    options.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
+                    options.LinkType = useParentChildLinking ? OpenTelemetryLinkType.AlwaysParentChild : OpenTelemetryLinkType.AlwaysLink;
+                })
                 .AddInMemoryExporter(exportedItems)
                 .Build();
             string baggageGuid = Guid.NewGuid().ToString();
             Baggage.SetBaggage("TestItem", baggageGuid);
             Assert.Equal(baggageGuid, Baggage.GetBaggage("TestItem"));
 
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             await Task.Delay(500);
 
             string queueName = $"{Guid.NewGuid()}";
@@ -186,24 +196,29 @@ public async Task TestPublisherAndConsumerActivityTagsAsync(bool useRoutingKeyAs
 
             await _channel.BasicCancelAsync(consumerTag);
             await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, queueName, exportedItems, true);
+            AssertActivityData(useRoutingKeyAsOperationName, useParentChildLinking, queueName, exportedItems, true);
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherWithPublicationAddressAndConsumerActivityTagsAsync(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(true, false)]
+        [InlineData(false, true)]
+        [InlineData(false, false)]
+        public async Task TestPublisherWithPublicationAddressAndConsumerActivityTagsAsync(bool useRoutingKeyAsOperationName, bool useParentChildLinking)
         {
             var exportedItems = new List<Activity>();
             using var tracer = Sdk.CreateTracerProviderBuilder()
-                .AddRabbitMQInstrumentation()
+                .AddRabbitMQInstrumentation(options =>
+                {
+                    options.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
+                    options.LinkType = useParentChildLinking ? OpenTelemetryLinkType.AlwaysParentChild : OpenTelemetryLinkType.AlwaysLink;
+                })
                 .AddInMemoryExporter(exportedItems)
                 .Build();
             string baggageGuid = Guid.NewGuid().ToString();
             Baggage.SetBaggage("TestItem", baggageGuid);
             Assert.Equal(baggageGuid, Baggage.GetBaggage("TestItem"));
 
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             await Task.Delay(500);
 
             string queueName = $"{Guid.NewGuid()}";
@@ -241,24 +256,29 @@ public async Task TestPublisherWithPublicationAddressAndConsumerActivityTagsAsyn
 
             await _channel.BasicCancelAsync(consumerTag);
             await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, queueName, exportedItems, true);
+            AssertActivityData(useRoutingKeyAsOperationName, useParentChildLinking, queueName, exportedItems, true);
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherWithCachedStringsAndConsumerActivityTagsAsync(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(true, false)]
+        [InlineData(false, true)]
+        [InlineData(false, false)]
+        public async Task TestPublisherWithCachedStringsAndConsumerActivityTagsAsync(bool useRoutingKeyAsOperationName, bool useParentChildLinking)
         {
             var exportedItems = new List<Activity>();
             using var tracer = Sdk.CreateTracerProviderBuilder()
-                .AddRabbitMQInstrumentation()
+                .AddRabbitMQInstrumentation(options =>
+                {
+                    options.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
+                    options.LinkType = useParentChildLinking ? OpenTelemetryLinkType.AlwaysParentChild : OpenTelemetryLinkType.AlwaysLink;
+                })
                 .AddInMemoryExporter(exportedItems)
                 .Build();
             string baggageGuid = Guid.NewGuid().ToString();
             Baggage.SetBaggage("TestItem", baggageGuid);
             Assert.Equal(baggageGuid, Baggage.GetBaggage("TestItem"));
 
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             await Task.Delay(500);
 
             string queueName = $"{Guid.NewGuid()}";
@@ -297,23 +317,28 @@ public async Task TestPublisherWithCachedStringsAndConsumerActivityTagsAsync(boo
 
             await _channel.BasicCancelAsync(consumerTag);
             await Task.Delay(500);
-            AssertActivityData(useRoutingKeyAsOperationName, queueName, exportedItems, true);
+            AssertActivityData(useRoutingKeyAsOperationName, useParentChildLinking, queueName, exportedItems, true);
         }
 
         [Theory]
-        [InlineData(true)]
-        [InlineData(false)]
-        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName)
+        [InlineData(true, true)]
+        [InlineData(true, false)]
+        [InlineData(false, true)]
+        [InlineData(false, false)]
+        public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOperationName, bool useParentChildLinking)
         {
             var exportedItems = new List<Activity>();
             using var tracer = Sdk.CreateTracerProviderBuilder()
-                .AddRabbitMQInstrumentation()
+                .AddRabbitMQInstrumentation(options =>
+                {
+                    options.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
+                    options.LinkType = useParentChildLinking ? OpenTelemetryLinkType.AlwaysParentChild : OpenTelemetryLinkType.AlwaysLink;
+                })
                 .AddInMemoryExporter(exportedItems)
                 .Build();
             string baggageGuid = Guid.NewGuid().ToString();
             Baggage.SetBaggage("TestItem", baggageGuid);
             Assert.Equal(baggageGuid, Baggage.GetBaggage("TestItem"));
-            RabbitMQActivitySource.UseRoutingKeyAsOperationName = useRoutingKeyAsOperationName;
             await Task.Delay(500);
             string queue = $"queue-{Guid.NewGuid()}";
             const string msg = "for basic.get";
@@ -331,15 +356,15 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
                 ok = await _channel.QueueDeclarePassiveAsync(queue);
                 Assert.Equal(0u, ok.MessageCount);
                 await Task.Delay(500);
-                AssertActivityData(useRoutingKeyAsOperationName, queue, exportedItems, false);
+                AssertActivityData(useRoutingKeyAsOperationName, useParentChildLinking, queue, exportedItems, false);
             }
             finally
             {
                 await _channel.QueueDeleteAsync(queue);
             }
         }
 
-        private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,
+        private void AssertActivityData(bool useRoutingKeyAsOperationName, bool useParentChildLinking, string queueName,
             List<Activity> activityList, bool isDeliver = false, string baggageGuid = null)
         {
             string childName = isDeliver ? "deliver" : "fetch";
@@ -359,11 +384,21 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
                 x.GetTagItem(RabbitMQActivitySource.MessagingDestinationRoutingKey) is string routingKeyTag &&
                 routingKeyTag == $"{queueName}");
             Activity receiveActivity = activities.Single(x =>
-                x.OperationName == (useRoutingKeyAsOperationName ? $"{childName} {queueName}" : childName) &&
-                x.Links.First().Context.TraceId == sendActivity.TraceId);
+                x.OperationName == (useRoutingKeyAsOperationName ? $"{childName} {queueName}" : childName));
             Assert.Equal(ActivityKind.Producer, sendActivity.Kind);
             Assert.Equal(ActivityKind.Consumer, receiveActivity.Kind);
-            Assert.Null(receiveActivity.ParentId);
+            if (useParentChildLinking)
+            {
+                Assert.Empty(receiveActivity.Links);
+                Assert.Equal(sendActivity.Id, receiveActivity.ParentId);
+                Assert.Equal(sendActivity.TraceId, receiveActivity.TraceId);
+            }
+            else
+            {
+                Assert.Equal(sendActivity.TraceId, receiveActivity.Links.Single().Context.TraceId);
+                Assert.Null(receiveActivity.ParentId);
+                Assert.NotEqual(sendActivity.TraceId, receiveActivity.TraceId);
+            }
             AssertStringTagNotNullOrEmpty(sendActivity, "network.peer.address");
             AssertStringTagNotNullOrEmpty(sendActivity, "network.local.address");
             AssertStringTagNotNullOrEmpty(sendActivity, "server.address");
@@ -385,6 +420,7 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
             AssertStringTagEquals(receiveActivity, RabbitMQActivitySource.MessagingOperationName, childName);
             AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessagingOperationType, "send");
             AssertStringTagEquals(sendActivity, RabbitMQActivitySource.MessagingOperationName, "publish");
+
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`1`	`+~static OpenTelemetry.Trace.OpenTelemetryExtensions.AddRabbitMQInstrumentation(this OpenTelemetry.Trace.TracerProviderBuilder builder, System.Action<RabbitMQ.Client.RabbitMQOpenTelemetryOptions> configure) -> OpenTelemetry.Trace.TracerProviderBuilder`
Original file line number	Diff line number	Diff line change
`@@ -8,16 +8,27 @@`
`8`	`8`
`9`	`9`	`namespace OpenTelemetry.Trace`
`10`	`10`	`{`
	`11`	`+`
`11`	`12`	`public static class OpenTelemetryExtensions`
`12`	`13`	`{`
`13`		`- public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProviderBuilder builder)`
	`14`	`+ public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProviderBuilder builder, Action<RabbitMQOpenTelemetryOptions> configure)`
`14`	`15`	`{`
	`16`	`+ var options = new RabbitMQOpenTelemetryOptions();`
	`17`	`+ configure?.Invoke(options);`
	`18`	`+ RabbitMQActivitySource.LinkType = options.LinkType;`
	`19`	`+ RabbitMQActivitySource.UseRoutingKeyAsOperationName = options.UseRoutingKeyAsOperationName;`
	`20`	`+`
`15`	`21`	`RabbitMQActivitySource.ContextExtractor = OpenTelemetryContextExtractor;`
`16`	`22`	`RabbitMQActivitySource.ContextInjector = OpenTelemetryContextInjector;`
`17`	`23`	`builder.AddSource("RabbitMQ.Client.*");`
`18`	`24`	`return builder;`
`19`	`25`	`}`
`20`	`26`
	`27`	`+ public static TracerProviderBuilder AddRabbitMQInstrumentation(this TracerProviderBuilder builder)`
	`28`	`+ {`
	`29`	`+ return AddRabbitMQInstrumentation(builder, null);`
	`30`	`+ }`
	`31`	`+`
`21`	`32`	`private static ActivityContext OpenTelemetryContextExtractor(IReadOnlyBasicProperties props)`
`22`	`33`	`{`
`23`	`34`	`// Extract the PropagationContext of the upstream parent from the message headers.`