Calculate timeSinceEnqueued for consumer spans

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

@@ -41,6 +41,10 @@ public RequestData(int version, Activity activity, ref ActivityTagsProcessor act
 
             TraceHelper.AddActivityLinksToProperties(activity, ref activityTagsProcessor.UnMappedTags);
             TraceHelper.AddPropertiesToTelemetry(Properties, ref activityTagsProcessor.UnMappedTags);
+            if (activity.Kind == ActivityKind.Consumer)
+            {
+                TraceHelper.AddEnqueuedTimeToMeasurements(activity, Measurements);
+            }
         }
 
         internal static bool IsSuccess(Activity activity, string? responseCode, OperationType operationType)

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

@@ -300,5 +300,55 @@ private static void AddTelemetryFromActivityEvents(Activity activity, TelemetryI
                 BaseData = exceptionData,
             };
         }
+
+        internal static void AddEnqueuedTimeToMeasurements(Activity activity, IDictionary<string, double> measurements)
+        {
+            if (activity.Links != null && activity.Links.Any())
+            {
+                if (TryGetAverageTimeInQueueForBatch(activity.Links, activity.StartTimeUtc, out long enqueuedTime))
+                {
+                    measurements["timeSinceEnqueued"] = enqueuedTime;
+                }
+            }
+        }
+
+        private static bool TryGetAverageTimeInQueueForBatch(IEnumerable<ActivityLink> links, DateTimeOffset requestStartTime, out long avgTimeInQueue)
+        {
+            avgTimeInQueue = 0;
+            var linksCount = 0;
+            long startEpochTime = requestStartTime.ToUnixTimeMilliseconds();
+            foreach (ActivityLink link in links)
+            {
+                if (!TryGetEnqueuedTime(link, out var msgEnqueuedTime))
+                {
+                    // instrumentation does not consistently report enqueued time, ignoring whole span
+                    return false;
+                }
+
+                avgTimeInQueue += Math.Max(startEpochTime - msgEnqueuedTime, 0);
+
+                linksCount++;
+            }
+
+            avgTimeInQueue /= linksCount;
+            return true;
+        }
+
+        private static bool TryGetEnqueuedTime(ActivityLink link, out long enqueuedTime)
+        {
+            enqueuedTime = 0;
+            if (link.Tags != null && link.Tags.Any())
+            {
+                foreach (var attribute in link.Tags)
+                {
+                    if (attribute.Key == "enqueuedTime")
+                    {
+                        return long.TryParse(attribute.Value?.ToString(), out enqueuedTime);
+                    }
+                }
+            }
+
+            return false;
+        }
     }
 }

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

@@ -2,9 +2,9 @@
 // Licensed under the MIT License.
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.Globalization;
-
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
 using Azure.Monitor.OpenTelemetry.Exporter.Models;
 
@@ -133,5 +133,100 @@ public void RequestDataContainsAzureNamespace()
             Assert.True(activityTagsProcessor.HasAzureNamespace);
             Assert.Equal("DemoAzureResource", requestData.Properties["az.namespace"]);
         }
+
+        [Fact]
+        public void RequestDataContainsTimeSinceEnqueuedForConsumerSpans()
+        {
+            using ActivitySource activitySource = new ActivitySource(ActivitySourceName);
+            List<ActivityLink>? links = new List<ActivityLink>();
+            long enqueued0 = DateTimeOffset.UtcNow.AddMilliseconds(-100).ToUnixTimeMilliseconds();
+            long enqueued1 = DateTimeOffset.UtcNow.AddMilliseconds(-200).ToUnixTimeMilliseconds();
+            long enqueued2 = DateTimeOffset.UtcNow.AddMilliseconds(-300).ToUnixTimeMilliseconds();
+
+            links.Add(AddActivityLink(enqueued0));
+            links.Add(AddActivityLink(enqueued1));
+            links.Add(AddActivityLink(enqueued2));
+
+            using var activity = activitySource.StartActivity("Activity", ActivityKind.Consumer, null, null, links);
+            Assert.NotNull(activity);
+            var activityTagsProcessor = TraceHelper.EnumerateActivityTags(activity);
+
+            var requestData = new RequestData(2, activity, ref activityTagsProcessor);
+
+            DateTimeOffset startTime = activity.StartTimeUtc;
+            var startTimeEpoch = startTime.ToUnixTimeMilliseconds();
+
+            long expectedTimeInQueue = ((startTimeEpoch - enqueued0) +
+                                      (startTimeEpoch - enqueued1) +
+                                      (startTimeEpoch - enqueued2)) / 3; // avg diff with request start time across links
+
+            Assert.True(requestData.Measurements.TryGetValue("timeSinceEnqueued", out var timeInQueue));
+
+            Assert.Equal(expectedTimeInQueue, timeInQueue);
+        }
+
+        [Fact]
+        public void RequestDataTimeSinceEnqueuedNegative()
+        {
+            using ActivitySource activitySource = new ActivitySource(ActivitySourceName);
+            List<ActivityLink>? links = new List<ActivityLink>();
+            long enqueued0 = DateTimeOffset.UtcNow.AddMilliseconds(-100).ToUnixTimeMilliseconds();
+            long enqueued1 = DateTimeOffset.UtcNow.AddMilliseconds(-200).ToUnixTimeMilliseconds();
+            long enqueued2 = DateTimeOffset.UtcNow.AddMilliseconds(300).ToUnixTimeMilliseconds(); // ignored
+
+            links.Add(AddActivityLink(enqueued0));
+            links.Add(AddActivityLink(enqueued1));
+            links.Add(AddActivityLink(enqueued2));
+
+            using var activity = activitySource.StartActivity("Activity", ActivityKind.Consumer, null, null, links);
+            Assert.NotNull(activity);
+            var activityTagsProcessor = TraceHelper.EnumerateActivityTags(activity);
+
+            var requestData = new RequestData(2, activity, ref activityTagsProcessor);
+
+            DateTimeOffset startTime = activity.StartTimeUtc;
+            var startTimeEpoch = startTime.ToUnixTimeMilliseconds();
+
+            long expectedTimeInQueue = ((startTimeEpoch - enqueued0) +
+                                      (startTimeEpoch - enqueued1)) / 3; // avg diff with request start time across links
+
+            Assert.True(requestData.Measurements.TryGetValue("timeSinceEnqueued", out var timeInQueue));
+
+            Assert.Equal(expectedTimeInQueue, timeInQueue);
+        }
+
+        [Fact]
+        public void RequestDataTimeSinceEnqueuedInvalidEmqueuedTime()
+        {
+            using ActivitySource activitySource = new ActivitySource(ActivitySourceName);
+            List<ActivityLink>? links = new List<ActivityLink>();
+
+            ActivityTagsCollection tags = new ActivityTagsCollection();
+            tags.Add("enqueuedTime", "Invalid");
+            var link = new ActivityLink(new ActivityContext(ActivityTraceId.CreateRandom(), ActivitySpanId.CreateRandom(), ActivityTraceFlags.None, null), tags);
+            links.Add(link);
+
+            using var activity = activitySource.StartActivity("Activity", ActivityKind.Consumer, null, null, links);
+            Assert.NotNull(activity);
+            var activityTagsProcessor = TraceHelper.EnumerateActivityTags(activity);
+
+            var requestData = new RequestData(2, activity, ref activityTagsProcessor);
+
+            DateTimeOffset startTime = activity.StartTimeUtc;
+            var startTimeEpoch = startTime.ToUnixTimeMilliseconds();
+
+            Assert.False(requestData.Measurements.TryGetValue("timeSinceEnqueued", out var timeInQueue));
+        }
+
+        private ActivityLink AddActivityLink(long enqueuedTime)
+        {
+            ActivityTagsCollection tags = new ActivityTagsCollection
+            {
+                { "enqueuedTime", enqueuedTime.ToString() }
+            };
+            var link = new ActivityLink(new ActivityContext(ActivityTraceId.CreateRandom(), ActivitySpanId.CreateRandom(), ActivityTraceFlags.None, null), tags);
+
+            return link;
+        }
     }
 }