[AzureMonitorExporter] Add support new Messaging semantics - Request/Dependency Telemetry

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Customizations/Models/RemoteDependencyData.cs

@@ -71,9 +71,10 @@ public RemoteDependencyData(int version, Activity activity, ref ActivityTagsProc
                     ResultCode = depInfo[2]?.ToString().Truncate(SchemaConstants.RemoteDependencyData_ResultCode_MaxLength);
                     break;
                 case OperationType.Messaging:
-                    depDataAndType = AzMonList.GetTagValues(ref activityTagsProcessor.MappedTags, SemanticConventions.AttributeMessagingUrl, SemanticConventions.AttributeMessagingSystem);
-                    Data = depDataAndType[0]?.ToString().Truncate(SchemaConstants.RemoteDependencyData_Data_MaxLength);
-                    Type = depDataAndType[1]?.ToString().Truncate(SchemaConstants.RemoteDependencyData_Type_MaxLength);
+                    var messagingSystem = AzMonList.GetTagValue(ref activityTagsProcessor.MappedTags, SemanticConventions.AttributeMessagingSystem);
+                    var messagingUrl = activityTagsProcessor.MappedTags.GetNewSchemaMessagingUrl(activity.Kind);
+                    Data = messagingUrl?.Truncate(SchemaConstants.RemoteDependencyData_Data_MaxLength);
+                    Type = messagingSystem?.ToString().Truncate(SchemaConstants.RemoteDependencyData_Type_MaxLength);
                     break;
             }
 

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Customizations/Models/RequestData.Schema.New.cs

@@ -3,6 +3,7 @@
 
 using System.Diagnostics;
 using System.Globalization;
+using System.Runtime.CompilerServices;
 using Azure.Core;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
 
@@ -24,12 +25,8 @@ public RequestData(int version, Activity activity, ref ActivityTagsProcessor act
         Duration = activity.Duration < SchemaConstants.RequestData_Duration_LessThanDays
             ? activity.Duration.ToString("c", CultureInfo.InvariantCulture)
             : SchemaConstants.Duration_MaxValue;
-        ResponseCode = AzMonList.GetTagValue(ref activityTagsProcessor.MappedTags, SemanticConventions.AttributeHttpResponseStatusCode)
-            ?.ToString().Truncate(SchemaConstants.RequestData_ResponseCode_MaxLength)
-            ?? "0";
-
+        ResponseCode = GetResponseCode(ref activityTagsProcessor.MappedTags);
         Success = IsSuccess(activity, ResponseCode, activityTagsProcessor.activityType);
-
         Url = url.Truncate(SchemaConstants.RequestData_Url_MaxLength);
         Properties = new ChangeTrackingDictionary<string, string>();
         Measurements = new ChangeTrackingDictionary<string, double>();
@@ -45,4 +42,12 @@ public RequestData(int version, Activity activity, ref ActivityTagsProcessor act
 
         TraceHelper.AddPropertiesToTelemetry(Properties, ref activityTagsProcessor.UnMappedTags);
     }
+
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    internal static string GetResponseCode(ref AzMonList mappedTags)
+    {
+        return AzMonList.GetTagValue(ref mappedTags, SemanticConventions.AttributeHttpResponseStatusCode)
+            ?.ToString().Truncate(SchemaConstants.RequestData_ResponseCode_MaxLength)
+            ?? "0";
+    }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Customizations/Models/RequestData.cs

@@ -3,6 +3,7 @@
 
 using System.Diagnostics;
 using System.Globalization;
+using System.Runtime.CompilerServices;
 using Azure.Core;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
 
@@ -20,7 +21,7 @@ public RequestData(int version, Activity activity, ref ActivityTagsProcessor act
                     url = activityTagsProcessor.MappedTags.GetRequestUrl();
                     break;
                 case OperationType.Messaging:
-                    url = AzMonList.GetTagValue(ref activityTagsProcessor.MappedTags, SemanticConventions.AttributeMessagingUrl)?.ToString();
+                    url = activityTagsProcessor.MappedTags.GetNewSchemaMessagingUrl(activity.Kind);
                     break;
             }
 
@@ -51,6 +52,7 @@ public RequestData(int version, Activity activity, ref ActivityTagsProcessor act
             TraceHelper.AddPropertiesToTelemetry(Properties, ref activityTagsProcessor.UnMappedTags);
         }
 
+        [MethodImpl(MethodImplOptions.AggressiveInlining)]
         internal static bool IsSuccess(Activity activity, string? responseCode, OperationType operationType)
         {
             if (operationType.HasFlag(OperationType.Http)

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/ActivityTagsProcessor.cs

@@ -61,10 +61,10 @@ internal struct ActivityTagsProcessor
             SemanticConventions.AttributeEndpointAddress,
             // required - Messaging
             SemanticConventions.AttributeMessagingSystem,
-            SemanticConventions.AttributeMessagingDestination,
-            SemanticConventions.AttributeMessagingDestinationKind,
-            SemanticConventions.AttributeMessagingTempDestination,
             SemanticConventions.AttributeMessagingUrl,
+            SemanticConventions.AttributeNetworkProtocolName,
+            SemanticConventions.AttributeMessagingDestinationName,
+            SemanticConventions.AttributeMessagingSourceName,
 
             // Others
             SemanticConventions.AttributeEnduserId

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/AzMonNewListExtensions.cs

@@ -2,7 +2,9 @@
 // Licensed under the MIT License.
 
 using System;
+using System.Diagnostics;
 using System.Runtime.CompilerServices;
+using Microsoft.Extensions.Primitives;
 
 namespace Azure.Monitor.OpenTelemetry.Exporter.Internals;
 
@@ -15,28 +17,85 @@ internal static class AzMonNewListExtensions
     {
         try
         {
-            var serverAddress = AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeServerAddress)?.ToString();
-            if (serverAddress != null)
+            var requestUrlTagObjects = AzMonList.GetTagValues(ref tagObjects, SemanticConventions.AttributeUrlScheme, SemanticConventions.AttributeServerAddress, SemanticConventions.AttributeServerPort, SemanticConventions.AttributeUrlPath, SemanticConventions.AttributeUrlQuery);
+
+            var scheme = requestUrlTagObjects[0]?.ToString() ?? string.Empty; // requestUrlTagObjects[0] => SemanticConventions.AttributeUrlScheme.
+            var host = requestUrlTagObjects[1]?.ToString() ?? string.Empty; // requestUrlTagObjects[1] => SemanticConventions.AttributeServerAddress.
+            var port = requestUrlTagObjects[2]?.ToString(); // requestUrlTagObjects[2] => SemanticConventions.AttributeServerPort.
+            port = port != null ? port = $":{port}" : string.Empty;
+            var path = requestUrlTagObjects[3]?.ToString() ?? string.Empty; // requestUrlTagObjects[3] => SemanticConventions.AttributeUrlPath.
+            var queryString = requestUrlTagObjects[4]?.ToString() ?? string.Empty; // requestUrlTagObjects[4] => SemanticConventions.AttributeUrlQuery.
+
+            var length = scheme.Length + Uri.SchemeDelimiter.Length + host.Length + port.Length + path.Length + queryString.Length;
+
+            var urlStringBuilder = new System.Text.StringBuilder(length)
+                .Append(scheme)
+                .Append(Uri.SchemeDelimiter)
+                .Append(host)
+                .Append(port)
+                .Append(path)
+                .Append(queryString);
+
+            return urlStringBuilder.ToString();
+        }
+        catch
+        {
+            // If URI building fails, there is no need to throw an exception. Instead, we can simply return null.
+        }
+
+        return null;
+    }
+
+    ///<summary>
+    /// Gets messaging url from activity tag objects.
+    ///</summary>
+    ///<example>
+    /// amqps://my.servicebus.windows.net/queueName.
+    ///</example>
+    internal static string? GetNewSchemaMessagingUrl(this AzMonList tagObjects, ActivityKind activityKind)
+    {
+        try
+        {
+            var host = AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeServerAddress)?.ToString()
+                        ?? AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeNetPeerName)?.ToString();
+            if (!string.IsNullOrEmpty(host))
             {
-                UriBuilder uriBuilder = new()
-                {
-                    Scheme = AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeUrlScheme)?.ToString(),
-                    Host = serverAddress,
-                    Path = AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeUrlPath)?.ToString(),
-                    Query = AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeUrlQuery)?.ToString()
-                };
+                object?[] messagingTagObjects;
+                string? sourceOrDestinationName = null;
 
-                if (int.TryParse(AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeServerPort)?.ToString(), out int port))
+                if (activityKind == ActivityKind.Consumer)
+                {
+                    messagingTagObjects = AzMonList.GetTagValues(ref tagObjects, SemanticConventions.AttributeNetworkProtocolName, SemanticConventions.AttributeMessagingSourceName);
+                    sourceOrDestinationName = messagingTagObjects[1]?.ToString(); // messagingTagObjects[1] => SemanticConventions.AttributeMessagingSourceName.
+                    if (sourceOrDestinationName == null)
+                    {
+                        sourceOrDestinationName = AzMonList.GetTagValue(ref tagObjects, SemanticConventions.AttributeMessagingDestinationName)?.ToString();
+                    }
+                }
+                else
                 {
-                    uriBuilder.Port = port;
+                    messagingTagObjects = AzMonList.GetTagValues(ref tagObjects, SemanticConventions.AttributeNetworkProtocolName, SemanticConventions.AttributeMessagingDestinationName);
+                    sourceOrDestinationName = messagingTagObjects[1]?.ToString(); // messagingTagObjects[1] => SemanticConventions.AttributeMessagingDestinationName.
                 }
 
-                return uriBuilder.Uri.AbsoluteUri;
+                var protocolName = messagingTagObjects[0]?.ToString() ?? string.Empty; // messagingTagObjects[0] => SemanticConventions.AttributeNetworkProtocolName.
+                sourceOrDestinationName ??= string.Empty;
+                var delimilerLength = 1;
+
+                var length = protocolName.Length + (protocolName?.Length > 0 ? Uri.SchemeDelimiter.Length : 0) + host!.Length + delimilerLength + sourceOrDestinationName.Length;
+
+                var messagingStringBuilder = new System.Text.StringBuilder(length)
+                    .Append(protocolName)
+                    .Append(string.IsNullOrEmpty(protocolName) ? null : Uri.SchemeDelimiter)
+                    .Append(host)
+                    .Append(string.IsNullOrEmpty(sourceOrDestinationName) ? "/" : $"/{sourceOrDestinationName}");
+
+                return messagingStringBuilder.ToString();
             }
         }
         catch
         {
-            // If URI building fails, there is no need to throw an exception. Instead, we can simply return null.
+            // If Messaging Url building fails, there is no need to throw an exception. Instead, we can simply return null.
         }
 
         return null;
@@ -88,4 +147,17 @@ private static bool IsDefaultPort(int port)
     {
         return port == 0 || port == 80 || port == 443;
     }
+
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    private static bool TryGetNonDefaultPort(string? stringport, out string? port)
+    {
+        port = stringport;
+
+        if (string.IsNullOrEmpty(stringport) || stringport == "80" || stringport == "443")
+        {
+            return false;
+        }
+
+        return true;
+    }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/src/Internals/SemanticConventions.cs

@@ -163,5 +163,10 @@ internal static class SemanticConventions
         public const string AttributeUrlQuery = "url.query";
         public const string AttributeUserAgentOriginal = "user_agent.original"; // replaces: "http.user_agent" (AttributeHttpUserAgent)
         public const string AttributeServerSocketAddress = "server.socket.address"; // replaces: "net.peer.ip" (AttributeNetPeerIp)
+
+        // Messaging v1.21.0 https://github.com/open-telemetry/opentelemetry-specification/blob/v1.21.0/specification/trace/semantic_conventions/messaging.md
+        public const string AttributeNetworkProtocolName = "network.protocol.name";
+        public const string AttributeMessagingDestinationName = "messaging.destination.name";
+        public const string AttributeMessagingSourceName = "messaging.source.name";
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.Exporter/tests/Azure.Monitor.OpenTelemetry.Exporter.Tests/AzMonNewListExtensionsTests.cs

@@ -1,8 +1,9 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 
+using System;
 using System.Collections.Generic;
-
+using System.Diagnostics;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
 using Xunit;
 
@@ -11,13 +12,11 @@ namespace Azure.Monitor.OpenTelemetry.Exporter.Tests
     public class AzMonNewListExtensionsTests
     {
         [Theory]
-        [InlineData("http", "example.com", "8080", "/search", "q=OpenTelemetry", "http://example.com:8080/search?q=OpenTelemetry")]
-        [InlineData(null, "example.com", "8080", "/search", "q=OpenTelemetry", "example.com:8080/search?q=OpenTelemetry")]
-        [InlineData("http", null, "8080", "/search", "q=OpenTelemetry", null)]
-        [InlineData("http", "example.com", null, "/search", "q=OpenTelemetry", "http://example.com/search?q=OpenTelemetry")]
-        [InlineData("http", "example.com", "8080", null, "q=OpenTelemetry", "http://example.com:8080/?q=OpenTelemetry")]
+        [InlineData("http", "example.com", "8080", "/search", "?q=OpenTelemetry", "http://example.com:8080/search?q=OpenTelemetry")]
+        [InlineData("http", "example.com", null, "/search", "?q=OpenTelemetry", "http://example.com/search?q=OpenTelemetry")]
+        [InlineData("http", "example.com", "8080", "/", "?q=OpenTelemetry", "http://example.com:8080/?q=OpenTelemetry")]
         [InlineData("http", "example.com", "8080", "/search", null, "http://example.com:8080/search")]
-        public void GetNewRequestUrl_ReturnsCorrectUrl(string urlScheme, string serverAddress, string serverPort, string urlPath, string urlQuery, string expectedUrl)
+        public void GetNewRequestUrl_ReturnsCorrectUrl(string urlScheme, string serverAddress, string? serverPort, string urlPath, string? urlQuery, string expectedUrl)
         {
             // Arrange
             AzMonList tagObjects = AzMonList.Initialize();
@@ -27,11 +26,85 @@ public void GetNewRequestUrl_ReturnsCorrectUrl(string urlScheme, string serverAd
             AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeUrlPath, urlPath));
             AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeUrlQuery, urlQuery));
 
+            // Validate the length with the same logic from code.
+            int colonLength = serverPort == null ? 0 : 1;
+            serverPort ??= string.Empty;
+            urlQuery ??= string.Empty;
+            var length = urlScheme.Length + Uri.SchemeDelimiter.Length + serverAddress.Length + serverPort.Length + colonLength + urlPath.Length + urlQuery.Length;
+
             // Act
             string? url = tagObjects.GetNewSchemaRequestUrl();
 
             // Assert
             Assert.Equal(expectedUrl, url);
+            Assert.Equal(length, url?.Length);
+        }
+
+        [Theory]
+        [InlineData("my.servicebus.windows.net", "amqps", "queueName", "amqps://my.servicebus.windows.net/queueName")]
+        [InlineData("my.servicebus.windows.net", "amqps", "", "amqps://my.servicebus.windows.net/")]
+        [InlineData("", "amqps", "queueName", null)]
+        [InlineData("my.servicebus.windows.net", "", "queueName", "my.servicebus.windows.net/queueName")]
+        [InlineData("my.servicebus.windows.net", null, null, "my.servicebus.windows.net/")]
+        [InlineData(null, "amqps", "queueName", null)]
+        public void GetNewSchemaMessagingUrl_ActivityKindProducer(string serverAddress, string protocolName, string destinationName, string expectedUrl)
+        {
+            // Arrange
+            AzMonList tagObjects = AzMonList.Initialize();
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeServerAddress, serverAddress));
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeNetworkProtocolName, protocolName));
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeMessagingDestinationName, destinationName));
+
+            // Act
+            var url = tagObjects.GetNewSchemaMessagingUrl(ActivityKind.Producer);
+
+            // Assert
+            Assert.Equal(expectedUrl, url);
+            if (url != null)
+            {
+                // Validate the length with the same logic from code.
+                var length = (protocolName?.Length ?? 0) + (protocolName?.Length > 0 ? Uri.SchemeDelimiter.Length : 0) + serverAddress!.Length + (destinationName?.Length ?? 0) + 1;
+                Assert.Equal(length, url.Length);
+            }
+        }
+
+        [Theory]
+        [InlineData("my.servicebus.windows.net", "amqps", "queueName", null, "amqps://my.servicebus.windows.net/queueName")]
+        [InlineData("my.servicebus.windows.net", "amqps", "", null, "amqps://my.servicebus.windows.net/")]
+        [InlineData("", "amqps", null, null, null)]
+        [InlineData("my.servicebus.windows.net", "", null, "destinationName", "my.servicebus.windows.net/destinationName")]
+        [InlineData(null, "amqps", null, "destinationName", null)]
+        [InlineData("my.servicebus.windows.net", "amqps", "sourceName", "destinationName", "amqps://my.servicebus.windows.net/sourceName")]
+        [InlineData("my.servicebus.windows.net", "amqps", null, null, "amqps://my.servicebus.windows.net/")]
+        [InlineData("my.servicebus.windows.net", "amqps", "", "", "amqps://my.servicebus.windows.net/")]
+        public void GetNewSchemaMessagingUrl_ActivityKindConsumer(string serverAddress, string protocolName, string sourceName, string destinationName, string expectedUrl)
+        {
+            // Arrange
+            AzMonList tagObjects = AzMonList.Initialize();
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeServerAddress, serverAddress));
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeNetworkProtocolName, protocolName));
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeMessagingDestinationName, destinationName));
+            AzMonList.Add(ref tagObjects, new KeyValuePair<string, object?>(SemanticConventions.AttributeMessagingSourceName, sourceName));
+
+            // Act
+            var result = tagObjects.GetNewSchemaMessagingUrl(ActivityKind.Consumer);
+
+            // Assert
+            Assert.Equal(expectedUrl, result);
+        }
+
+        [Fact]
+        public void GetNewSchemaMessagingUrl_NullTagObjects_ReturnsNull()
+        {
+            // Arrange
+            AzMonList tagObjects = AzMonList.Initialize();
+            var activityKind = ActivityKind.Consumer;
+
+            // Act
+            var result = tagObjects.GetNewSchemaMessagingUrl(activityKind);
+
+            // Assert
+            Assert.Null(result);
         }
 
         [Theory]