Enhanced support for EventHub and minor convention changes for Azure SDK

CHANGELOG.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## VNext
+- [Support new conventions for EventHubs from Azure.Messaging.EventHubs and processor.](https://github.com/microsoft/ApplicationInsights-dotnet/pull/1674)
 - [Adding a flag to DependencyTrackingTelemetryModule to enable/disable collection of SQL Command text.](https://github.com/microsoft/ApplicationInsights-dotnet/pull/1514)
    Collecting SQL Command Text will now be opt-in, so this value will default to false. This is a change from the current behavior on .NET Core. To see how to collect SQL Command Text see here for details: https://docs.microsoft.com/en-us/azure/azure-monitor/app/asp-net-dependencies#advanced-sql-tracking-to-get-full-sql-query
 - [change references to log4net to version 2.0.8](https://github.com/microsoft/ApplicationInsights-dotnet/issues/1675)

WEB/Src/DependencyCollector/Shared.Tests/Implementation/EventHubsDiagnosticListenerTests.cs

@@ -85,7 +85,7 @@ public void EventHubsSuccessfulSendIsHandled()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Send", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.AreEqual(sendActivity.ParentSpanId.ToHexString(), telemetry.Context.Operation.ParentId);
@@ -125,7 +125,7 @@ public void EventHubsSuccessfulReceiveIsHandled()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Receive", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.AreEqual(sendActivity.ParentSpanId.ToHexString(), telemetry.Context.Operation.ParentId);
@@ -165,7 +165,7 @@ public void EventHubsSuccessfulReceiveShortNameIsHandled()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Receive", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.AreEqual(sendActivity.ParentSpanId.ToHexString(), telemetry.Context.Operation.ParentId);
@@ -205,7 +205,7 @@ public void EventHubsSuccessfulSendShortNameIsHandled()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Send", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.AreEqual(sendActivity.ParentSpanId.ToHexString(), telemetry.Context.Operation.ParentId);
@@ -246,7 +246,7 @@ public void EventHubsSuccessfulSendIsHandledW3COff()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Send", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.AreEqual(parentActivity.Id, telemetry.Context.Operation.ParentId);
@@ -281,7 +281,7 @@ public void EventHubsSuccessfulSendIsHandledWithoutParent()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Send", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.IsNull(telemetry.Context.Operation.ParentId);
@@ -315,7 +315,7 @@ public void EventHubsSuccessfulSendIsHandledWithExternalParent()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Send", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsTrue(telemetry.Success.Value);
 
                 Assert.AreEqual("parent", telemetry.Context.Operation.ParentId);
@@ -350,7 +350,7 @@ public void EventHubsFailedSendIsHandled()
                 Assert.IsNotNull(telemetry);
                 Assert.AreEqual("Send", telemetry.Name);
                 Assert.AreEqual(RemoteDependencyConstants.AzureEventHubs, telemetry.Type);
-                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ | ehname", telemetry.Target);
+                Assert.AreEqual("sb://eventhubname.servicebus.windows.net/ehname", telemetry.Target);
                 Assert.IsFalse(telemetry.Success.Value);
 
                 Assert.AreEqual(sendActivity.ParentSpanId.ToHexString(), telemetry.Context.Operation.ParentId);

WEB/Src/DependencyCollector/Shared/Implementation/AzureSdk/AzureSdkDiagnosticsEventHandler.cs

@@ -15,6 +15,10 @@
 
     internal class AzureSdkDiagnosticsEventHandler : DiagnosticsEventHandlerBase
     {
+#if NET45
+        private static readonly DateTimeOffset EpochStart = new DateTime(1970, 1, 1, 0, 0, 0, 0, System.DateTimeKind.Utc);
+#endif
+
         private readonly ObjectInstanceBasedOperationHolder<OperationTelemetry> operationHolder = new ObjectInstanceBasedOperationHolder<OperationTelemetry>();
 
         // fetchers must not be reused between sources
@@ -48,62 +52,43 @@ public override void OnEvent(KeyValuePair<string, object> evnt, DiagnosticListen
                         }
                     }
 
+                    string type = this.GetType(currentActivity);
+
                     if (telemetry == null)
                     {
-                        string dependencyType = RemoteDependencyConstants.InProc;
-                        foreach (var tag in currentActivity.Tags)
-                        {
-                            if (tag.Key == "kind")
-                            {
-                                if (tag.Value == "internal")
-                                {
-                                    break;
-                                }
-
-                                dependencyType = string.Empty;
-                            }
-
-                            if (tag.Key.StartsWith("http.", StringComparison.Ordinal))
-                            {
-                                dependencyType = RemoteDependencyConstants.HTTP;
-                                break;
-                            }
-
-                            if (tag.Key == "component" && tag.Value == "eventhubs")
-                            {
-                                dependencyType = RemoteDependencyConstants.AzureEventHubs;
-                                break;
-                            }
-                        }
+                        telemetry = new DependencyTelemetry { Type = type };
+                    }
 
-                        telemetry = new DependencyTelemetry { Type = dependencyType };
+                    if (type != null && type.EndsWith(RemoteDependencyConstants.AzureEventHubs, StringComparison.Ordinal))
+                    {
+                        this.SetEventHubsProperties(currentActivity, telemetry);
                     }
 
                     if (this.linksPropertyFetcher.Fetch(evnt.Value) is IEnumerable<Activity> activityLinks)
                     {
                         this.PopulateLinks(activityLinks, telemetry);
+
+                        if (telemetry is RequestTelemetry request &&
+                            TryGetAverageTimeInQueueForBatch(activityLinks, currentActivity.StartTimeUtc, out long enqueuedTime))
+                        {
+                            request.Metrics["timeSinceEnqueued"] = enqueuedTime;
+                        }
                     }
 
                     this.operationHolder.Store(currentActivity, Tuple.Create(telemetry, /* isCustomCreated: */ false));
                 }
                 else if (evnt.Key.EndsWith(".Stop", StringComparison.Ordinal))
                 {
                     var telemetry = this.operationHolder.Get(currentActivity).Item1;
+
                     this.SetCommonProperties(evnt.Key, evnt.Value, currentActivity, telemetry);
 
-                    if (telemetry is DependencyTelemetry dependency)
+                    if (telemetry is DependencyTelemetry dependency && dependency.Type == RemoteDependencyConstants.HTTP)
                     {
-                        if (dependency.Type == RemoteDependencyConstants.HTTP)
-                        {
-                            this.SetHttpProperties(currentActivity, dependency);
-                            if (evnt.Value != null)
-                            {
-                                dependency.SetOperationDetail(evnt.Value.GetType().FullName, evnt.Value);
-                            }
-                        }
-                        else if (dependency.Type == RemoteDependencyConstants.AzureEventHubs)
+                        this.SetHttpProperties(currentActivity, dependency);
+                        if (evnt.Value != null)
                         {
-                            this.SetEventHubsProperties(currentActivity, dependency);
+                            dependency.SetOperationDetail(evnt.Value.GetType().FullName, evnt.Value);
                         }
                     }
 
@@ -157,6 +142,106 @@ protected override bool IsOperationSuccessful(string eventName, object eventPayl
             return true;
         }
 
+        private static bool TryGetAverageTimeInQueueForBatch(IEnumerable<Activity> links, DateTimeOffset requestStartTime, out long avgTimeInQueue)
+        {
+            avgTimeInQueue = 0;
+            int linksCount = 0;
+            foreach (var link in links)
+            {
+                if (!TryGetEnqueuedTime(link, out var msgEnqueuedTime))
+                {
+                    // instrumentation does not consistently report enqueued time, ignoring whole span
+                    return false;
+                }
+
+                long startEpochTime = 0;
+#if NET45
+                startEpochTime = (long)(requestStartTime - EpochStart).TotalMilliseconds;
+#else
+                startEpochTime = requestStartTime.ToUnixTimeMilliseconds();
+#endif
+                avgTimeInQueue += Math.Max(startEpochTime - msgEnqueuedTime, 0);
+                linksCount++;
+            }
+
+            if (linksCount == 0)
+            {
+                return false;
+            }
+
+            avgTimeInQueue /= linksCount;
+            return true;
+        }
+
+        private static bool TryGetEnqueuedTime(Activity link, out long enqueuedTime)
+        {
+            enqueuedTime = 0;
+            foreach (var attribute in link.Tags)
+            {
+                if (attribute.Key == "enqueuedTime")
+                {
+                    if (attribute.Value is string strValue)
+                    {
+                        return long.TryParse(strValue, out enqueuedTime);
+                    }
+                }
+            }
+
+            return false;
+        }
+
+        private string GetType(Activity currentActivity)
+        {
+            string kind = RemoteDependencyConstants.InProc;
+            string component = null;
+            foreach (var tag in currentActivity.Tags)
+            {
+                if (tag.Key.StartsWith("http.", StringComparison.Ordinal))
+                {
+                    return RemoteDependencyConstants.HTTP;
+                }
+
+                switch (tag.Key)
+                {
+                    case "kind":
+                        switch (tag.Value)
+                        {
+                            case "internal":
+                                break;
+                            case "producer":
+                                kind = RemoteDependencyConstants.QueueMessage;
+                                break;
+                            default:
+                                kind = null;
+                                break;
+                        }
+
+                        break;
+
+                    case "component":
+                    case "az.namespace":
+                        component = tag.Value;
+                        break;
+                }
+            }
+
+            if (component == "eventhubs" || component == "Microsoft.EventHub")
+            {
+                return kind == null
+                    ? RemoteDependencyConstants.AzureEventHubs
+                    : string.Concat(kind, " | ", RemoteDependencyConstants.AzureEventHubs);
+            }
+
+            if (component != null)
+            {
+                return kind == null
+                    ? component
+                    : string.Concat(kind, " | ", component);
+            }
+
+            return kind ?? string.Empty;
+        }
+
         private void SetHttpProperties(Activity activity, DependencyTelemetry dependency)
         {
             string method = null;
@@ -205,7 +290,7 @@ private void SetHttpProperties(Activity activity, DependencyTelemetry dependency
             }
         }
 
-        private void SetEventHubsProperties(Activity activity, DependencyTelemetry dependency)
+        private void SetEventHubsProperties(Activity activity, OperationTelemetry telemetry)
         {
             string endpoint = null;
             string queueName = null;
@@ -222,9 +307,29 @@ private void SetEventHubsProperties(Activity activity, DependencyTelemetry depen
                 }
             }
 
+            if (endpoint == null || queueName == null)
+            {
+                return;
+            }
+
             // Target uniquely identifies the resource, we use both: queueName and endpoint 
             // with schema used for SQL-dependencies
-            dependency.Target = string.Concat(endpoint, " | ", queueName);
+            string separator = "/";
+            if (endpoint.EndsWith(separator, StringComparison.Ordinal))
+            {
+                separator = string.Empty;
+            }
+
+            string eventHubInfo = string.Concat(endpoint, separator, queueName);
+
+            if (telemetry is DependencyTelemetry dependency)
+            {
+                dependency.Target = eventHubInfo;
+            }
+            else if (telemetry is RequestTelemetry request)
+            {
+                request.Source = eventHubInfo;
+            }
         }
 
         private void PopulateLinks(IEnumerable<Activity> links, OperationTelemetry telemetry)

WEB/Src/DependencyCollector/Shared/Implementation/EventHandlers/DiagnosticsEventHandlerBase.cs

@@ -40,7 +40,11 @@ public virtual bool IsEventEnabled(string evnt, object arg1, object arg2)
 
         protected void SetCommonProperties(string eventName, object eventPayload, Activity activity, OperationTelemetry telemetry)
         {
-            telemetry.Name = this.GetOperationName(eventName, eventPayload, activity);
+            if (string.IsNullOrEmpty(telemetry.Name))
+            {
+                telemetry.Name = this.GetOperationName(eventName, eventPayload, activity);
+            }
+
             telemetry.Duration = activity.Duration;
             telemetry.Timestamp = activity.StartTimeUtc;
 

WEB/Src/DependencyCollector/Shared/Implementation/EventHandlers/EventHubsDiagnosticsEventHandler.cs

@@ -58,9 +58,21 @@ private void OnDependency(string name, object payload, Activity activity)
 
             // Target uniquely identifies the resource, we use both: queueName and endpoint 
             // with schema used for SQL-dependencies
-            telemetry.Target = string.Join(" | ", endpoint, queueName);
+            if (endpoint == null || queueName == null)
+            {
+                return;
+            }
+
+            // Target uniquely identifies the resource, we use both: queueName and endpoint 
+            // with schema used for SQL-dependencies
+            string separator = "/";
+            if (endpoint.EndsWith(separator, StringComparison.Ordinal))
+            {
+                separator = string.Empty;
+            }
 
+            telemetry.Target = string.Concat(endpoint, separator, queueName);
             this.TelemetryClient.TrackDependency(telemetry);
         }
     }
-}
+}

WEB/Src/DependencyCollector/Shared/Implementation/RemoteDependencyConstants.cs

@@ -15,6 +15,7 @@ internal static class RemoteDependencyConstants
         public const string AzureIotHub = "Azure IoT Hub";
         public const string AzureSearch = "Azure Search";
         public const string InProc = "InProc";
+        public const string QueueMessage = "Queue Message";
 
         public const string WcfService = "WCF Service";
         public const string WebService = "Web Service";