Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions .azure-pipelines/ultimate-pipeline.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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
Expand Down
9 changes: 9 additions & 0 deletions docker-compose.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
3 changes: 2 additions & 1 deletion docker/eventgrid-emulator-config.json
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
{
"Topics": {
"samples-eventgrid-topic": []
"samples-eventgrid-topic": [],
"samples-azure-functions-eventgrid-topic": []
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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;
Expand All @@ -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;
Expand All @@ -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)
Expand All @@ -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)
{
Expand Down Expand Up @@ -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)
{
Expand All @@ -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;
}
}
}

Expand Down
Original file line number Diff line number Diff line change
@@ -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
Loading
Loading