Merge pull request #1717 from iinuwa/feat/update-otel-conventions

lukebakken · web-flow · commit 0496126e37db · 2024-12-13T19:59:39.000Z
More updates for the current OTel (OpenTelemetry) conventions
diff --git a/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs b/projects/RabbitMQ.Client/Impl/Channel.BasicPublish.cs
@@ -62,7 +62,7 @@ await MaybeEnforceFlowControlAsync(cancellationToken)
                 var cmd = new BasicPublish(exchange, routingKey, mandatory, default);
 
                 using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners
-                    ? RabbitMQActivitySource.Send(routingKey, exchange, body.Length)
+                    ? RabbitMQActivitySource.BasicPublish(routingKey, exchange, body.Length)
                     : default;
 
                 ulong publishSequenceNumber = 0;
@@ -117,7 +117,7 @@ await MaybeEnforceFlowControlAsync(cancellationToken)
                 var cmd = new BasicPublishMemory(exchange.Bytes, routingKey.Bytes, mandatory, default);
 
                 using Activity? sendActivity = RabbitMQActivitySource.PublisherHasListeners
-                    ? RabbitMQActivitySource.Send(routingKey.Value, exchange.Value, body.Length)
+                    ? RabbitMQActivitySource.BasicPublish(routingKey.Value, exchange.Value, body.Length)
                     : default;
 
                 ulong publishSequenceNumber = 0;
diff --git a/projects/RabbitMQ.Client/Impl/Channel.cs b/projects/RabbitMQ.Client/Impl/Channel.cs
@@ -918,10 +918,10 @@ await ModelSendAsync(in method, k.CancellationToken)
                 BasicGetResult? result = await k;
 
                 using Activity? activity = result != null
-                    ? RabbitMQActivitySource.Receive(result.RoutingKey,
+                    ? RabbitMQActivitySource.BasicGet(result.RoutingKey,
                         result.Exchange,
                         result.DeliveryTag, result.BasicProperties, result.Body.Length)
-                    : RabbitMQActivitySource.ReceiveEmpty(queue);
+                    : RabbitMQActivitySource.BasicGetEmpty(queue);
 
                 activity?.SetStartTime(k.StartTime);
 
diff --git a/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs b/projects/RabbitMQ.Client/Impl/RabbitMQActivitySource.cs
@@ -15,6 +15,11 @@ public static class RabbitMQActivitySource
         // https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes
         internal const string MessageId = "messaging.message.id";
         internal const string MessageConversationId = "messaging.message.conversation_id";
+        internal const string MessagingOperationName = "messaging.operation.name";
+        internal const string MessagingOperationNameBasicDeliver = "deliver";
+        internal const string MessagingOperationNameBasicGet = "fetch";
+        internal const string MessagingOperationNameBasicGetEmpty = "fetch (empty)";
+        internal const string MessagingOperationNameBasicPublish = "publish";
         internal const string MessagingOperationType = "messaging.operation.type";
         internal const string MessagingOperationTypeSend = "send";
         internal const string MessagingOperationTypeProcess = "process";
@@ -56,7 +61,7 @@ public static class RabbitMQActivitySource
             new KeyValuePair<string, object?>(ProtocolVersion, "0.9.1")
         };
 
-        internal static Activity? Send(string routingKey, string exchange, int bodySize,
+        internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize,
             ActivityContext linkedContext = default)
         {
             if (!s_publisherSource.HasListeners())
@@ -66,41 +71,42 @@ public static class RabbitMQActivitySource
 
             Activity? activity = linkedContext == default
                 ? s_publisherSource.StartRabbitMQActivity(
-                    UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeSend}" : MessagingOperationTypeSend,
+                    UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicPublish} {routingKey}" : MessagingOperationNameBasicPublish,
                     ActivityKind.Producer)
                 : s_publisherSource.StartLinkedRabbitMQActivity(
-                    UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeSend}" : MessagingOperationTypeSend,
+                    UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicPublish} {routingKey}" : MessagingOperationNameBasicPublish,
                     ActivityKind.Producer, linkedContext);
             if (activity != null && activity.IsAllDataRequested)
             {
-                PopulateMessagingTags(MessagingOperationTypeSend, routingKey, exchange, 0, bodySize, activity);
+                PopulateMessagingTags(MessagingOperationTypeSend, MessagingOperationNameBasicPublish, routingKey, exchange, 0, bodySize, activity);
             }
 
             return activity;
 
         }
 
-        internal static Activity? ReceiveEmpty(string queue)
+        internal static Activity? BasicGetEmpty(string queue)
         {
             if (!s_subscriberSource.HasListeners())
             {
                 return null;
             }
 
             Activity? activity = s_subscriberSource.StartRabbitMQActivity(
-                UseRoutingKeyAsOperationName ? $"{queue} {MessagingOperationTypeReceive}" : MessagingOperationTypeReceive,
+                UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicGetEmpty} {queue}" : MessagingOperationNameBasicGetEmpty,
                 ActivityKind.Consumer);
             if (activity != null && activity.IsAllDataRequested)
             {
                 activity
                     .SetTag(MessagingOperationType, MessagingOperationTypeReceive)
+                    .SetTag(MessagingOperationName, MessagingOperationNameBasicGetEmpty)
                     .SetTag(MessagingDestination, "amq.default");
             }
 
             return activity;
         }
 
-        internal static Activity? Receive(string routingKey, string exchange, ulong deliveryTag,
+        internal static Activity? BasicGet(string routingKey, string exchange, ulong deliveryTag,
             IReadOnlyBasicProperties readOnlyBasicProperties, int bodySize)
         {
             if (!s_subscriberSource.HasListeners())
@@ -110,11 +116,11 @@ public static class RabbitMQActivitySource
 
             // Extract the PropagationContext of the upstream parent from the message headers.
             Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(
-                UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeReceive}" : MessagingOperationTypeReceive, ActivityKind.Consumer,
+                UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicGet} {routingKey}" : MessagingOperationNameBasicGet, ActivityKind.Consumer,
                 ContextExtractor(readOnlyBasicProperties));
             if (activity != null && activity.IsAllDataRequested)
             {
-                PopulateMessagingTags(MessagingOperationTypeReceive, routingKey, exchange, deliveryTag, readOnlyBasicProperties,
+                PopulateMessagingTags(MessagingOperationTypeReceive, MessagingOperationNameBasicGet, routingKey, exchange, deliveryTag, readOnlyBasicProperties,
                     bodySize, activity);
             }
 
@@ -131,11 +137,11 @@ public static class RabbitMQActivitySource
 
             // Extract the PropagationContext of the upstream parent from the message headers.
             Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(
-                UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeProcess}" : MessagingOperationTypeProcess,
+                UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicDeliver} {routingKey}" : MessagingOperationNameBasicDeliver,
                 ActivityKind.Consumer, ContextExtractor(basicProperties));
             if (activity != null && activity.IsAllDataRequested)
             {
-                PopulateMessagingTags(MessagingOperationTypeProcess, routingKey, exchange,
+                PopulateMessagingTags(MessagingOperationTypeProcess, MessagingOperationNameBasicDeliver, routingKey, exchange,
                     deliveryTag, basicProperties, bodySize, activity);
             }
 
@@ -157,10 +163,10 @@ public static class RabbitMQActivitySource
                 ?.Start();
         }
 
-        private static void PopulateMessagingTags(string operation, string routingKey, string exchange,
+        private static void PopulateMessagingTags(string operationType, string operationName, string routingKey, string exchange,
             ulong deliveryTag, IReadOnlyBasicProperties readOnlyBasicProperties, int bodySize, Activity activity)
         {
-            PopulateMessagingTags(operation, routingKey, exchange, deliveryTag, bodySize, activity);
+            PopulateMessagingTags(operationType, operationName, routingKey, exchange, deliveryTag, bodySize, activity);
 
             if (!string.IsNullOrEmpty(readOnlyBasicProperties.CorrelationId))
             {
@@ -173,11 +179,12 @@ private static void PopulateMessagingTags(string operation, string routingKey, s
             }
         }
 
-        private static void PopulateMessagingTags(string operation, string routingKey, string exchange,
+        private static void PopulateMessagingTags(string operationType, string operationName, string routingKey, string exchange,
             ulong deliveryTag, int bodySize, Activity activity)
         {
             activity
-                .SetTag(MessagingOperationType, operation)
+                .SetTag(MessagingOperationType, operationType)
+                .SetTag(MessagingOperationName, operationName)
                 .SetTag(MessagingDestination, string.IsNullOrEmpty(exchange) ? "amq.default" : exchange)
                 .SetTag(MessagingDestinationRoutingKey, routingKey)
                 .SetTag(MessagingBodySize, bodySize);
diff --git a/projects/Test/SequentialIntegration/TestActivitySource.cs b/projects/Test/SequentialIntegration/TestActivitySource.cs
@@ -402,7 +402,7 @@ private static ActivityListener StartActivityListener(List<Activity> activities)
         private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,
             List<Activity> activityList, bool isDeliver = false)
         {
-            string childName = isDeliver ? "process" : "receive";
+            string childName = isDeliver ? "deliver" : "fetch";
             Activity[] activities = activityList.ToArray();
             Assert.NotEmpty(activities);
 
@@ -418,11 +418,11 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
             }
 
             Activity sendActivity = activities.First(x =>
-                x.OperationName == (useRoutingKeyAsOperationName ? $"{queueName} send" : "send") &&
+                x.OperationName == (useRoutingKeyAsOperationName ? $"publish {queueName}" : "publish") &&
                 x.GetTagItem(RabbitMQActivitySource.MessagingDestinationRoutingKey) is string routingKeyTag &&
                 routingKeyTag == $"{queueName}");
             Activity receiveActivity = activities.Single(x =>
-                x.OperationName == (useRoutingKeyAsOperationName ? $"{queueName} {childName}" : $"{childName}") &&
+                x.OperationName == (useRoutingKeyAsOperationName ? $"{childName} {queueName}" : childName) &&
                 x.Links.First().Context.TraceId == sendActivity.TraceId);
             Assert.Equal(ActivityKind.Producer, sendActivity.Kind);
             Assert.Equal(ActivityKind.Consumer, receiveActivity.Kind);
diff --git a/projects/Test/SequentialIntegration/TestOpenTelemetry.cs b/projects/Test/SequentialIntegration/TestOpenTelemetry.cs
@@ -342,7 +342,8 @@ public async Task TestPublisherAndBasicGetActivityTags(bool useRoutingKeyAsOpera
         private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueName,
             List<Activity> activityList, bool isDeliver = false, string baggageGuid = null)
         {
-            string childName = isDeliver ? "process" : "receive";
+            string childName = isDeliver ? "deliver" : "fetch";
+            string childType = isDeliver ? "process" : "receive";
             Activity[] activities = activityList.ToArray();
             Assert.NotEmpty(activities);
             foreach (var item in activities)
@@ -354,11 +355,11 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
             }
 
             Activity sendActivity = activities.First(x =>
-                x.OperationName == (useRoutingKeyAsOperationName ? $"{queueName} send" : "send") &&
+                x.OperationName == (useRoutingKeyAsOperationName ? $"publish {queueName}" : "publish") &&
                 x.GetTagItem(RabbitMQActivitySource.MessagingDestinationRoutingKey) is string routingKeyTag &&
                 routingKeyTag == $"{queueName}");
             Activity receiveActivity = activities.Single(x =>
-                x.OperationName == (useRoutingKeyAsOperationName ? $"{queueName} {childName}" : $"{childName}") &&
+                x.OperationName == (useRoutingKeyAsOperationName ? $"{childName} {queueName}" : childName) &&
                 x.Links.First().Context.TraceId == sendActivity.TraceId);
             Assert.Equal(ActivityKind.Producer, sendActivity.Kind);
             Assert.Equal(ActivityKind.Consumer, receiveActivity.Kind);
@@ -380,6 +381,10 @@ private void AssertActivityData(bool useRoutingKeyAsOperationName, string queueN
             AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingEnvelopeSize);
             AssertIntTagGreaterThanZero(sendActivity, RabbitMQActivitySource.MessagingBodySize);
             AssertIntTagGreaterThanZero(receiveActivity, RabbitMQActivitySource.MessagingBodySize);
+            AssertStringTagEquals(receiveActivity, RabbitMQActivitySource.MessagingOperationType, childType);
+            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
`@@ -15,6 +15,11 @@ public static class RabbitMQActivitySource`
`15`	`15`	`// https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/#messaging-attributes`
`16`	`16`	`internal const string MessageId = "messaging.message.id";`
`17`	`17`	`internal const string MessageConversationId = "messaging.message.conversation_id";`
	`18`	`+ internal const string MessagingOperationName = "messaging.operation.name";`
	`19`	`+ internal const string MessagingOperationNameBasicDeliver = "deliver";`
	`20`	`+ internal const string MessagingOperationNameBasicGet = "fetch";`
	`21`	`+ internal const string MessagingOperationNameBasicGetEmpty = "fetch (empty)";`
	`22`	`+ internal const string MessagingOperationNameBasicPublish = "publish";`
`18`	`23`	`internal const string MessagingOperationType = "messaging.operation.type";`
`19`	`24`	`internal const string MessagingOperationTypeSend = "send";`
`20`	`25`	`internal const string MessagingOperationTypeProcess = "process";`
`@@ -56,7 +61,7 @@ public static class RabbitMQActivitySource`
`56`	`61`	`new KeyValuePair<string, object?>(ProtocolVersion, "0.9.1")`
`57`	`62`	`};`
`58`	`63`
`59`		`- internal static Activity? Send(string routingKey, string exchange, int bodySize,`
	`64`	`+ internal static Activity? BasicPublish(string routingKey, string exchange, int bodySize,`
`60`	`65`	`ActivityContext linkedContext = default)`
`61`	`66`	`{`
`62`	`67`	`if (!s_publisherSource.HasListeners())`
`@@ -66,41 +71,42 @@ public static class RabbitMQActivitySource`
`66`	`71`
`67`	`72`	`Activity? activity = linkedContext == default`
`68`	`73`	`? s_publisherSource.StartRabbitMQActivity(`
`69`		`- UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeSend}" : MessagingOperationTypeSend,`
	`74`	`+ UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicPublish} {routingKey}" : MessagingOperationNameBasicPublish,`
`70`	`75`	`ActivityKind.Producer)`
`71`	`76`	`: s_publisherSource.StartLinkedRabbitMQActivity(`
`72`		`- UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeSend}" : MessagingOperationTypeSend,`
	`77`	`+ UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicPublish} {routingKey}" : MessagingOperationNameBasicPublish,`
`73`	`78`	`ActivityKind.Producer, linkedContext);`
`74`	`79`	`if (activity != null && activity.IsAllDataRequested)`
`75`	`80`	`{`
`76`		`- PopulateMessagingTags(MessagingOperationTypeSend, routingKey, exchange, 0, bodySize, activity);`
	`81`	`+ PopulateMessagingTags(MessagingOperationTypeSend, MessagingOperationNameBasicPublish, routingKey, exchange, 0, bodySize, activity);`
`77`	`82`	`}`
`78`	`83`
`79`	`84`	`return activity;`
`80`	`85`
`81`	`86`	`}`
`82`	`87`
`83`		`- internal static Activity? ReceiveEmpty(string queue)`
	`88`	`+ internal static Activity? BasicGetEmpty(string queue)`
`84`	`89`	`{`
`85`	`90`	`if (!s_subscriberSource.HasListeners())`
`86`	`91`	`{`
`87`	`92`	`return null;`
`88`	`93`	`}`
`89`	`94`
`90`	`95`	`Activity? activity = s_subscriberSource.StartRabbitMQActivity(`
`91`		`- UseRoutingKeyAsOperationName ? $"{queue} {MessagingOperationTypeReceive}" : MessagingOperationTypeReceive,`
	`96`	`+ UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicGetEmpty} {queue}" : MessagingOperationNameBasicGetEmpty,`
`92`	`97`	`ActivityKind.Consumer);`
`93`	`98`	`if (activity != null && activity.IsAllDataRequested)`
`94`	`99`	`{`
`95`	`100`	`activity`
`96`	`101`	`.SetTag(MessagingOperationType, MessagingOperationTypeReceive)`
	`102`	`+ .SetTag(MessagingOperationName, MessagingOperationNameBasicGetEmpty)`
`97`	`103`	`.SetTag(MessagingDestination, "amq.default");`
`98`	`104`	`}`
`99`	`105`
`100`	`106`	`return activity;`
`101`	`107`	`}`
`102`	`108`
`103`		`- internal static Activity? Receive(string routingKey, string exchange, ulong deliveryTag,`
	`109`	`+ internal static Activity? BasicGet(string routingKey, string exchange, ulong deliveryTag,`
`104`	`110`	`IReadOnlyBasicProperties readOnlyBasicProperties, int bodySize)`
`105`	`111`	`{`
`106`	`112`	`if (!s_subscriberSource.HasListeners())`
`@@ -110,11 +116,11 @@ public static class RabbitMQActivitySource`
`110`	`116`
`111`	`117`	`// Extract the PropagationContext of the upstream parent from the message headers.`
`112`	`118`	`Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(`
`113`		`- UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeReceive}" : MessagingOperationTypeReceive, ActivityKind.Consumer,`
	`119`	`+ UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicGet} {routingKey}" : MessagingOperationNameBasicGet, ActivityKind.Consumer,`
`114`	`120`	`ContextExtractor(readOnlyBasicProperties));`
`115`	`121`	`if (activity != null && activity.IsAllDataRequested)`
`116`	`122`	`{`
`117`		`- PopulateMessagingTags(MessagingOperationTypeReceive, routingKey, exchange, deliveryTag, readOnlyBasicProperties,`
	`123`	`+ PopulateMessagingTags(MessagingOperationTypeReceive, MessagingOperationNameBasicGet, routingKey, exchange, deliveryTag, readOnlyBasicProperties,`
`118`	`124`	`bodySize, activity);`
`119`	`125`	`}`
`120`	`126`
`@@ -131,11 +137,11 @@ public static class RabbitMQActivitySource`
`131`	`137`
`132`	`138`	`// Extract the PropagationContext of the upstream parent from the message headers.`
`133`	`139`	`Activity? activity = s_subscriberSource.StartLinkedRabbitMQActivity(`
`134`		`- UseRoutingKeyAsOperationName ? $"{routingKey} {MessagingOperationTypeProcess}" : MessagingOperationTypeProcess,`
	`140`	`+ UseRoutingKeyAsOperationName ? $"{MessagingOperationNameBasicDeliver} {routingKey}" : MessagingOperationNameBasicDeliver,`
`135`	`141`	`ActivityKind.Consumer, ContextExtractor(basicProperties));`
`136`	`142`	`if (activity != null && activity.IsAllDataRequested)`
`137`	`143`	`{`
`138`		`- PopulateMessagingTags(MessagingOperationTypeProcess, routingKey, exchange,`
	`144`	`+ PopulateMessagingTags(MessagingOperationTypeProcess, MessagingOperationNameBasicDeliver, routingKey, exchange,`
`139`	`145`	`deliveryTag, basicProperties, bodySize, activity);`
`140`	`146`	`}`
`141`	`147`
`@@ -157,10 +163,10 @@ public static class RabbitMQActivitySource`
`157`	`163`	`?.Start();`
`158`	`164`	`}`
`159`	`165`
`160`		`- private static void PopulateMessagingTags(string operation, string routingKey, string exchange,`
	`166`	`+ private static void PopulateMessagingTags(string operationType, string operationName, string routingKey, string exchange,`
`161`	`167`	`ulong deliveryTag, IReadOnlyBasicProperties readOnlyBasicProperties, int bodySize, Activity activity)`
`162`	`168`	`{`
`163`		`- PopulateMessagingTags(operation, routingKey, exchange, deliveryTag, bodySize, activity);`
	`169`	`+ PopulateMessagingTags(operationType, operationName, routingKey, exchange, deliveryTag, bodySize, activity);`
`164`	`170`
`165`	`171`	`if (!string.IsNullOrEmpty(readOnlyBasicProperties.CorrelationId))`
`166`	`172`	`{`
`@@ -173,11 +179,12 @@ private static void PopulateMessagingTags(string operation, string routingKey, s`
`173`	`179`	`}`
`174`	`180`	`}`
`175`	`181`
`176`		`- private static void PopulateMessagingTags(string operation, string routingKey, string exchange,`
	`182`	`+ private static void PopulateMessagingTags(string operationType, string operationName, string routingKey, string exchange,`
`177`	`183`	`ulong deliveryTag, int bodySize, Activity activity)`
`178`	`184`	`{`
`179`	`185`	`activity`
`180`		`- .SetTag(MessagingOperationType, operation)`
	`186`	`+ .SetTag(MessagingOperationType, operationType)`
	`187`	`+ .SetTag(MessagingOperationName, operationName)`
`181`	`188`	`.SetTag(MessagingDestination, string.IsNullOrEmpty(exchange) ? "amq.default" : exchange)`
`182`	`189`	`.SetTag(MessagingDestinationRoutingKey, routingKey)`
`183`	`190`	`.SetTag(MessagingBodySize, bodySize);`