[Azure Event Grid] Add Azure Functions support

Author: pablomartinezbernardo

.azure-pipelines/ultimate-pipeline.yml

@@ -2412,6 +2412,9 @@ stages:
         SampleName: $(IntegrationTestSampleName)
         DD_LOGGER_DD_API_KEY: $(ddApiKey)
         COMPOSE_PROFILES: group$(dockerGroup)
+        # Event Grid emulator (group2) pushes to the Functions host webhook; in CI the func host runs inside
+        # the IntegrationTests container, reachable by sibling containers as "integrationtests".
+        EVENTGRID_WEBHOOK_HOST: integrationtests
       displayName: docker-compose build IntegrationTests and run StartDependencies (Group $(dockerGroup))
       retryCountOnTaskFailure: 5
 
@@ -2431,6 +2434,7 @@ stages:
         DD_LOGGER_DD_API_KEY: $(ddApiKey)
         baseImage: $(baseImage) # for interpolation in the docker-compose file
         COMPOSE_PROFILES: group$(dockerGroup)
+        EVENTGRID_WEBHOOK_HOST: integrationtests
 
     - script: docker-compose -f docker-compose.yml -p $(DockerComposeProjectName)-g$(dockerGroup) logs
       displayName: docker-compose logs

docker-compose.yml

@@ -280,6 +280,14 @@ services:
       - "127.0.0.1:6500:6500"
     volumes:
       - ./docker/eventgrid-emulator-config.json:/app/appsettings.json:ro
+    environment:
+      # Override the topic's subscriber webhook (ASP.NET array config). The Azure Functions host that
+      # receives the pushed event is reachable at a different host depending on where it runs: inside the
+      # IntegrationTests container in CI (EVENTGRID_WEBHOOK_HOST=integrationtests) vs on the Docker host
+      # locally (host.docker.internal). The emulator adds the aeg-event-type header and forwards the CloudEvent.
+      - Topics__samples-azure-functions-eventgrid-topic__0=http://${EVENTGRID_WEBHOOK_HOST:-host.docker.internal}:7071/runtime/webhooks/eventgrid?functionName=EventGridTrigger
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
 
   cosmosdb-emulator:
     image: mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-preview@sha256:54d7bc334494c50cea867c270880671a7db080626a9732832b34c0d69342f9b0
@@ -465,6 +473,7 @@ services:
       - ASB_CONNECTION_STRING=Endpoint=sb://azureservicebus-emulator:5672;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;
       - EVENTHUBS_CONNECTION_STRING=Endpoint=sb://azure-eventhubs-emulator:5672;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=SAS_KEY_VALUE;UseDevelopmentEmulator=true;
       - EVENTGRID_TOPIC_ENDPOINT=http://azure-eventgrid-emulator:6500/samples-eventgrid-topic/api/events
+      - EVENTGRID_AZURE_FUNCTIONS_TOPIC_ENDPOINT=http://azure-eventgrid-emulator:6500/samples-azure-functions-eventgrid-topic/api/events
       - COSMOSDB_ENDPOINT=https://cosmosdb-emulator:8081
       - TEST_AGENT_HOST=test-agent
       - CONTAINER_HOSTNAME=http://integrationtests

docker/eventgrid-emulator-config.json

@@ -1,5 +1,6 @@
 {
   "Topics": {
-    "samples-eventgrid-topic": []
+    "samples-eventgrid-topic": [],
+    "samples-azure-functions-eventgrid-topic": []
   }
 }

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/Functions/AzureFunctionsCommon.cs

@@ -8,7 +8,6 @@
 using System;
 using System.Collections;
 using System.Collections.Generic;
-using System.Diagnostics.CodeAnalysis;
 using System.Linq;
 using Datadog.Trace.ClrProfiler.AutoInstrumentation.Azure.Shared;
 using Datadog.Trace.ClrProfiler.AutoInstrumentation.Proxy;
@@ -20,9 +19,6 @@
 using Datadog.Trace.PlatformHelpers;
 using Datadog.Trace.Propagators;
 using Datadog.Trace.Tagging;
-using Datadog.Trace.Util;
-using Datadog.Trace.Util.Json;
-using Datadog.Trace.Vendors.Newtonsoft.Json;
 using Datadog.Trace.Vendors.Serilog.Events;
 
 #nullable enable
@@ -314,6 +310,12 @@ _ when type.StartsWith("eventGrid", StringComparison.OrdinalIgnoreCase) => "Even
                         case "EventHub" when tracer.CurrentTraceSettings.Settings.IsIntegrationEnabled(IntegrationId.AzureEventHubs):
                             extractedContext = ExtractPropagatedContextFromMessaging(functionContext, "Properties", "PropertiesArray").MergeBaggageInto(Baggage.Current);
                             break;
+
+                        case "EventGrid" when tracer.CurrentTraceSettings.Settings.IsIntegrationEnabled(IntegrationId.AzureEventGrid, defaultValue: false):
+                        {
+                            extractedContext = EventGridFunctionsCommon.CreateReceiveSpanContext(tracer, functionContext, entry.Key as string, Baggage.Current);
+                            break;
+                        }
                     }
 
                     break;
@@ -483,23 +485,14 @@ private static PropagationContext ExtractPropagatedContextFromHttp<T>(T function
 
             try
             {
-                object? feature = null;
-                foreach (var keyValuePair in functionContext.Features)
-                {
-                    if (keyValuePair.Key.FullName?.Equals("Microsoft.Azure.Functions.Worker.Context.Features.IFunctionBindingsFeature") == true)
-                    {
-                        feature = keyValuePair.Value;
-                        break;
-                    }
-                }
-
-                if (feature is null || !feature.TryDuckCast<FunctionBindingsFeatureStruct>(out var bindingFeature))
+                var bindingsFeature = FunctionBindingsCommon.GetBindingsFeature(functionContext);
+                if (bindingsFeature is null)
                 {
                     return default;
                 }
 
-                if (bindingFeature.InputData is null
-                 || !bindingFeature.InputData.TryGetValue(bindingName!, out var requestDataObject)
+                if (bindingsFeature.Value.InputData is null
+                 || !bindingsFeature.Value.InputData.TryGetValue(bindingName!, out var requestDataObject)
                  || requestDataObject is null)
                 {
                     return default;
@@ -524,7 +517,7 @@ internal static PropagationContext ExtractPropagatedContextFromMessaging<T>(T co
         {
             try
             {
-                var bindingsFeature = GetFeatureFromContext<T, FunctionBindingsFeatureStruct>(context, "Microsoft.Azure.Functions.Worker.Context.Features.IFunctionBindingsFeature");
+                var bindingsFeature = FunctionBindingsCommon.GetBindingsFeature(context);
                 if (bindingsFeature == null)
                 {
                     return default;
@@ -535,7 +528,7 @@ internal static PropagationContext ExtractPropagatedContextFromMessaging<T>(T co
 
                 // Extract from single message properties
                 if (triggerMetadata?.TryGetValue(singlePropertyKey, out var singlePropsObj) == true &&
-                    TryParseJson<Dictionary<string, object>>(singlePropsObj, out var singleProps) && singleProps != null)
+                    FunctionBindingsCommon.TryParseJson<Dictionary<string, object>>(singlePropsObj, out var singleProps))
                 {
                     var singleContext = Shared.AzureMessagingCommon.ExtractContext(singleProps);
                     if (singleContext.SpanContext != null)
@@ -546,7 +539,7 @@ internal static PropagationContext ExtractPropagatedContextFromMessaging<T>(T co
 
                 // Extract from batch properties array
                 if (triggerMetadata?.TryGetValue(batchPropertyKey, out var arrayPropsObj) == true &&
-                    TryParseJson<Dictionary<string, object>[]>(arrayPropsObj, out var propsArray) && propsArray != null)
+                    FunctionBindingsCommon.TryParseJson<Dictionary<string, object>[]>(arrayPropsObj, out var propsArray))
                 {
                     foreach (var props in propsArray)
                     {
@@ -580,27 +573,6 @@ internal static PropagationContext ExtractPropagatedContextFromMessaging<T>(T co
             }
         }
 
-        private static bool TryParseJson<T>(object? jsonObj, [NotNullWhen(true)] out T? result)
-            where T : class
-        {
-            result = null;
-            if (jsonObj is not string jsonString)
-            {
-                return false;
-            }
-
-            try
-            {
-                result = JsonHelper.DeserializeObject<T>(jsonString);
-                return result != null;
-            }
-            catch (Exception ex)
-            {
-                Log.Debug(ex, "Failed to parse JSON: {Json}", jsonString);
-                return false;
-            }
-        }
-
         // Checks if all SpanContexts are identical (ignores baggage)
         private static bool AreAllSpanContextsIdentical(List<PropagationContext> contexts)
         {
@@ -615,26 +587,6 @@ private static bool AreAllSpanContextsIdentical(List<PropagationContext> context
                 ctx.SpanContext.TraceId128 == first!.TraceId128 &&
                 ctx.SpanContext.SpanId == first.SpanId);
         }
-
-        private static TFeature? GetFeatureFromContext<T, TFeature>(T context, string featureTypeName)
-            where T : IFunctionContext
-            where TFeature : struct
-        {
-            if (context.Features == null)
-            {
-                return null;
-            }
-
-            foreach (var kvp in context.Features)
-            {
-                if (kvp.Key.FullName == featureTypeName)
-                {
-                    return kvp.Value?.TryDuckCast<TFeature>(out var feature) == true ? feature : null;
-                }
-            }
-
-            return null;
-        }
     }
 }
 

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/Functions/EventGridFunctionsCommon.cs

@@ -0,0 +1,193 @@
+// <copyright file="EventGridFunctionsCommon.cs" company="Datadog">
+// Unless explicitly stated otherwise all files in this repository are licensed under the Apache 2 License.
+// This product includes software developed at Datadog (https://www.datadoghq.com/). Copyright 2017 Datadog, Inc.
+// </copyright>
+
+#if !NETFRAMEWORK
+
+using System;
+using System.Collections.Generic;
+using Datadog.Trace.ClrProfiler.AutoInstrumentation.Azure.Shared;
+using Datadog.Trace.Configuration;
+using Datadog.Trace.Configuration.Schema;
+using Datadog.Trace.Logging;
+using Datadog.Trace.Propagators;
+using Datadog.Trace.Tagging;
+using Datadog.Trace.Util;
+
+#nullable enable
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.Azure.Functions
+{
+    internal static class EventGridFunctionsCommon
+    {
+        private static readonly IDatadogLogger Log = DatadogLogging.GetLoggerFor(typeof(EventGridFunctionsCommon));
+
+        /// <summary>
+        /// Creates an <c>azure_eventgrid.receive</c> consumer span that links to the producers'
+        /// <c>azure_eventgrid.send</c> spans (when W3C contexts were extracted from the CloudEvents),
+        /// and returns a <see cref="PropagationContext"/> pointing at the receive span so the
+        /// function-invoke span is parented under it. This mirrors the Event Hubs / Service Bus
+        /// receive-span topology (new trace, span-linked to the producer).
+        /// </summary>
+        internal static PropagationContext CreateReceiveSpanContext<T>(Tracer tracer, T context, string? bindingName, Baggage destinationBaggage)
+            where T : IFunctionContext
+        {
+            var cloudEvents = GetCloudEvents(context, bindingName);
+            var producerContexts = ExtractPropagatedContexts(cloudEvents);
+
+            // As with Service Bus and Event Hubs batches, use the first extracted producer
+            // context as the source of ambient baggage.
+            if (producerContexts.Count > 0)
+            {
+                producerContexts[0].MergeBaggageInto(destinationBaggage);
+            }
+
+            return CreateReceiveSpan(tracer, cloudEvents, producerContexts);
+        }
+
+        internal static List<PropagationContext> ExtractPropagatedContexts(Dictionary<string, object>[] cloudEvents)
+        {
+            var extractedContexts = new List<PropagationContext>();
+
+            try
+            {
+                if (cloudEvents.Length == 0)
+                {
+                    return extractedContexts;
+                }
+
+                var uniqueSpanContexts = new HashSet<SpanContext>(new SpanContextComparer());
+
+                foreach (var cloudEventProps in cloudEvents)
+                {
+                    // Extract W3C trace context and baggage from CloudEvent extension attributes.
+                    // These were injected by EventGridCommon.InjectW3CContext() on the publisher side.
+                    var extractedContext = AzureMessagingCommon.ExtractContext(cloudEventProps);
+                    if (extractedContext.SpanContext is { } spanContext && uniqueSpanContexts.Add(spanContext))
+                    {
+                        extractedContexts.Add(extractedContext);
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error extracting propagated context from EventGrid binding");
+            }
+
+            return extractedContexts;
+        }
+
+        internal static List<SpanLink>? CreateSpanLinks(List<PropagationContext> producerContexts, bool linksEnabled)
+        {
+            if (!linksEnabled || producerContexts.Count == 0)
+            {
+                return null;
+            }
+
+            var links = new List<SpanLink>(producerContexts.Count);
+            foreach (var producerContext in producerContexts)
+            {
+                if (producerContext.SpanContext is { } producerSpanContext)
+                {
+                    links.Add(new SpanLink(producerSpanContext));
+                }
+            }
+
+            return links;
+        }
+
+        /// <summary>
+        /// Reads the CloudEvent JSON object or array for the given binding from <c>InputData</c>.
+        /// The full CloudEvent JSON (including extension attributes) lives in InputData, not
+        /// TriggerMetadata (which only contains <c>{"data": ...}</c> for Event Grid).
+        /// </summary>
+        internal static Dictionary<string, object>[] GetCloudEvents<T>(T context, string? bindingName)
+            where T : IFunctionContext
+        {
+            if (StringUtil.IsNullOrEmpty(bindingName))
+            {
+                return [];
+            }
+
+            var bindingsFeature = FunctionBindingsCommon.GetBindingsFeature(context);
+            if (bindingsFeature is null)
+            {
+                return [];
+            }
+
+            if (bindingsFeature.Value.InputData is null
+             || !bindingsFeature.Value.InputData.TryGetValue(bindingName!, out var inputDataObj)
+             || inputDataObj is not string inputDataJson)
+            {
+                return [];
+            }
+
+            var firstNonWhitespaceIndex = 0;
+            while (firstNonWhitespaceIndex < inputDataJson.Length && char.IsWhiteSpace(inputDataJson[firstNonWhitespaceIndex]))
+            {
+                firstNonWhitespaceIndex++;
+            }
+
+            if (firstNonWhitespaceIndex == inputDataJson.Length)
+            {
+                return [];
+            }
+
+            if (inputDataJson[firstNonWhitespaceIndex] == '[')
+            {
+                return FunctionBindingsCommon.TryParseJson<Dictionary<string, object>[]>(inputDataJson, out var cloudEventBatch) ? cloudEventBatch : [];
+            }
+
+            return FunctionBindingsCommon.TryParseJson<Dictionary<string, object>>(inputDataJson, out var cloudEventProps) ? [cloudEventProps] : [];
+        }
+
+        private static PropagationContext CreateReceiveSpan(Tracer tracer, Dictionary<string, object>[] cloudEvents, List<PropagationContext> producerContexts)
+        {
+            try
+            {
+                var tags = tracer.CurrentTraceSettings.Schema.Messaging.CreateAzureEventGridTags(SpanKinds.Consumer);
+                tags.MessagingOperation = "receive";
+
+                var links = CreateSpanLinks(producerContexts, tracer.Settings.AzureEventGridBatchLinksEnabled);
+                var (serviceName, serviceNameSource) = tracer.CurrentTraceSettings.Schema.Messaging.GetServiceNameMetadata(MessagingSchema.ServiceType.AzureEventGrid);
+
+                using var scope = tracer.StartActiveInternal(
+                    "azure_eventgrid.receive",
+                    parent: SpanContext.None,
+                    tags: tags,
+                    serviceName: serviceName,
+                    serviceNameSource: serviceNameSource,
+                    links: links);
+
+                var span = scope.Span;
+                span.Type = SpanTypes.Queue;
+                span.ResourceName = "eventgrid";
+
+                if (cloudEvents.Length == 1)
+                {
+                    var cloudEventProps = cloudEvents[0];
+                    if (cloudEventProps.TryGetValue("id", out var idObj) && idObj is string id && id.Length > 0)
+                    {
+                        span.SetTag(Tags.MessagingMessageId, id);
+                    }
+                }
+                else if (cloudEvents.Length > 1)
+                {
+                    tags.MessagingBatchMessageCount = cloudEvents.Length.ToString();
+                }
+
+                tracer.TracerManager.Telemetry.IntegrationGeneratedSpan(IntegrationId.AzureEventGrid);
+
+                return new PropagationContext(span.Context, baggage: null);
+            }
+            catch (Exception ex)
+            {
+                Log.Error(ex, "Error creating Azure Event Grid receive span");
+                return producerContexts.Count == 1 ? producerContexts[0] : default;
+            }
+        }
+    }
+}
+
+#endif