Implement optimizations for all integrations

kr-igor · kr-igor · commit adeec8fff58b · 2026-04-13T16:12:58.000-05:00
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/ContextPropagation.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/ContextPropagation.cs
@@ -64,7 +64,9 @@ public static void InjectTraceIntoData<TRecordRequest>(Tracer tracer, TRecordReq
                 if (dataStreamsManager != null && dataStreamsManager.IsEnabled)
                 {
                     var payloadSize = jsonData?.Count > 0 && record.Data != null ? record.Data.Length : 0;
-                    var edgeTags = new[] { "direction:out", $"topic:{streamName}", "type:kinesis" };
+                    var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(
+                        new KinesisEdgeTagCacheKey(streamName!, isConsume: false),
+                        static k => ["direction:out", $"topic:{k.StreamName}", "type:kinesis"]);
                     scope.Span.SetDataStreamsCheckpoint(
                         dataStreamsManager,
                         CheckpointKind.Produce,
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/GetRecordsAsyncIntegration.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/GetRecordsAsyncIntegration.cs
@@ -61,7 +61,9 @@ internal static TResponse OnAsyncMethodEnd<TTarget, TResponse>(TTarget instance,
                 var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager;
                 if (dataStreamsManager is { IsEnabled: true })
                 {
-                    var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:kinesis" };
+                    var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(
+                        new KinesisEdgeTagCacheKey((string)state.State, isConsume: true),
+                        static k => ["direction:in", $"topic:{k.StreamName}", "type:kinesis"]);
                     foreach (var o in response.Records)
                     {
                         var record = o.DuckCast<IRecord>();
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/GetRecordsIntegration.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/GetRecordsIntegration.cs
@@ -78,7 +78,9 @@ internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTar
                 var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager;
                 if (dataStreamsManager is { IsEnabled: true })
                 {
-                    var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:kinesis" };
+                    var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(
+                        new KinesisEdgeTagCacheKey((string)state.State, isConsume: true),
+                        static k => ["direction:in", $"topic:{k.StreamName}", "type:kinesis"]);
                     foreach (var o in response.Records)
                     {
                         var record = o.DuckCast<IRecord>();
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/KinesisEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/KinesisEdgeTagCacheKey.cs
@@ -0,0 +1,45 @@
+// <copyright file="KinesisEdgeTagCacheKey.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>
+
+#nullable enable
+
+using System;
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.Kinesis;
+
+/// <summary>
+/// Value-type cache key for Kinesis edge tags. Using a named struct avoids boxing and
+/// is compatible with all supported target frameworks.
+/// <see cref="IsConsume"/> distinguishes produce (direction:out) from consume (direction:in)
+/// so that both directions share a single cache type without key collision.
+/// </summary>
+internal readonly struct KinesisEdgeTagCacheKey : IEquatable<KinesisEdgeTagCacheKey>
+{
+    public readonly string StreamName;
+    public readonly bool IsConsume;
+
+    public KinesisEdgeTagCacheKey(string streamName, bool isConsume)
+    {
+        StreamName = streamName;
+        IsConsume = isConsume;
+    }
+
+    public bool Equals(KinesisEdgeTagCacheKey other)
+        => StreamName == other.StreamName && IsConsume == other.IsConsume;
+
+    public override bool Equals(object? obj)
+        => obj is KinesisEdgeTagCacheKey other && Equals(other);
+
+    public override int GetHashCode()
+    {
+        unchecked
+        {
+            int hash = 17;
+            hash = (hash * 31) + (StreamName?.GetHashCode() ?? 0);
+            hash = (hash * 31) + IsConsume.GetHashCode();
+            return hash;
+        }
+    }
+}
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SNS/AwsSnsHandlerCommon.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SNS/AwsSnsHandlerCommon.cs
@@ -37,8 +37,11 @@ public static CallTargetState BeforePublish<TPublishRequest>(TPublishRequest req
         if (scope?.Span.Context is { } context && !string.IsNullOrEmpty(topicName))
         {
             var dataStreamsManager = tracer.TracerManager.DataStreamsManager;
-            // avoid allocation if edgeTags are not going to be used
-            var edgeTags = dataStreamsManager is { IsEnabled: true } ? ["direction:out", $"topic:{topicName}", "type:sns"] : Array.Empty<string>();
+            var edgeTags = dataStreamsManager is { IsEnabled: true }
+                               ? dataStreamsManager.GetOrCreateEdgeTags(
+                                   new SnsEdgeTagCacheKey(topicName!),
+                                   static k => ["direction:out", $"topic:{k.TopicName}", "type:sns"])
+                               : Array.Empty<string>();
 
             if (sendType == SendType.SingleMessage)
             {
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SNS/SnsEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SNS/SnsEdgeTagCacheKey.cs
@@ -0,0 +1,33 @@
+// <copyright file="SnsEdgeTagCacheKey.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>
+
+#nullable enable
+
+using System;
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.SNS;
+
+/// <summary>
+/// Value-type cache key for SNS produce edge tags. Using a named struct avoids boxing and
+/// is compatible with all supported target frameworks.
+/// </summary>
+internal readonly struct SnsEdgeTagCacheKey : IEquatable<SnsEdgeTagCacheKey>
+{
+    public readonly string TopicName;
+
+    public SnsEdgeTagCacheKey(string topicName)
+    {
+        TopicName = topicName;
+    }
+
+    public bool Equals(SnsEdgeTagCacheKey other)
+        => TopicName == other.TopicName;
+
+    public override bool Equals(object? obj)
+        => obj is SnsEdgeTagCacheKey other && Equals(other);
+
+    public override int GetHashCode()
+        => TopicName?.GetHashCode() ?? 0;
+}
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SQS/AwsSqsHandlerCommon.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SQS/AwsSqsHandlerCommon.cs
@@ -66,7 +66,9 @@ private static void InjectForSingleMessage<TSendMessageRequest>(Tracer tracer, T
         var dataStreamsManager = tracer.TracerManager.DataStreamsManager;
         if (dataStreamsManager != null && dataStreamsManager.IsEnabled)
         {
-            var edgeTags = new[] { "direction:out", $"topic:{queueName}", "type:sqs" };
+            var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(
+                new SqsEdgeTagCacheKey(queueName, isConsume: false),
+                static k => ["direction:out", $"topic:{k.QueueName}", "type:sqs"]);
             scope.Span.SetDataStreamsCheckpoint(dataStreamsManager, CheckpointKind.Produce, edgeTags, payloadSizeBytes: 0, timeInQueueMs: 0);
         }
 
@@ -81,14 +83,18 @@ private static void InjectForBatch<TSendMessageBatchRequest>(Tracer tracer, TSen
             return;
         }
 
-        var edgeTags = new[] { "direction:out", $"topic:{queueName}", "type:sqs" };
+        var dataStreamsManager = tracer.TracerManager.DataStreamsManager;
+        var edgeTags = dataStreamsManager is { IsEnabled: true }
+                           ? dataStreamsManager.GetOrCreateEdgeTags(
+                               new SqsEdgeTagCacheKey(queueName, isConsume: false),
+                               static k => ["direction:out", $"topic:{k.QueueName}", "type:sqs"])
+                           : new[] { "direction:out", $"topic:{queueName}", "type:sqs" };
         foreach (var e in requestProxy.Entries)
         {
             var entry = e.DuckCast<IContainsMessageAttributes>();
             if (entry != null)
             {
                 // this has no effect if DSM is disabled
-                var dataStreamsManager = tracer.TracerManager.DataStreamsManager;
                 scope.Span.SetDataStreamsCheckpoint(dataStreamsManager, CheckpointKind.Produce, edgeTags, payloadSizeBytes: 0, timeInQueueMs: 0);
                 // this needs to be done for context propagation even when DSM is disabled
                 // (when DSM is enabled, it injects the pathway context on top of the trace context)
@@ -148,7 +154,9 @@ internal static TResponse AfterReceive<TResponse>(TResponse response, Exception?
             var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager;
             if (dataStreamsManager is { IsEnabled: true })
             {
-                var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:sqs" };
+                var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(
+                    new SqsEdgeTagCacheKey((string)state.State, isConsume: true),
+                    static k => ["direction:in", $"topic:{k.QueueName}", "type:sqs"]);
                 foreach (var o in response.Messages)
                 {
                     var message = o.DuckCast<IMessage>();
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SQS/SqsEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SQS/SqsEdgeTagCacheKey.cs
@@ -0,0 +1,45 @@
+// <copyright file="SqsEdgeTagCacheKey.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>
+
+#nullable enable
+
+using System;
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.AWS.SQS;
+
+/// <summary>
+/// Value-type cache key for SQS edge tags. Using a named struct avoids boxing and
+/// is compatible with all supported target frameworks.
+/// <see cref="IsConsume"/> distinguishes produce (direction:out) from consume (direction:in)
+/// so that both directions share a single cache type without key collision.
+/// </summary>
+internal readonly struct SqsEdgeTagCacheKey : IEquatable<SqsEdgeTagCacheKey>
+{
+    public readonly string QueueName;
+    public readonly bool IsConsume;
+
+    public SqsEdgeTagCacheKey(string queueName, bool isConsume)
+    {
+        QueueName = queueName;
+        IsConsume = isConsume;
+    }
+
+    public bool Equals(SqsEdgeTagCacheKey other)
+        => QueueName == other.QueueName && IsConsume == other.IsConsume;
+
+    public override bool Equals(object? obj)
+        => obj is SqsEdgeTagCacheKey other && Equals(other);
+
+    public override int GetHashCode()
+    {
+        unchecked
+        {
+            int hash = 17;
+            hash = (hash * 31) + (QueueName?.GetHashCode() ?? 0);
+            hash = (hash * 31) + IsConsume.GetHashCode();
+            return hash;
+        }
+    }
+}
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/ServiceBus/ProcessMessageIntegration.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/ServiceBus/ProcessMessageIntegration.cs
@@ -34,6 +34,9 @@ public sealed class ProcessMessageIntegration
     {
         private static readonly IDatadogLogger Log = DatadogLogging.GetLoggerFor(typeof(ProcessMessageIntegration));
 
+        // Used when the entity path is unknown — direction:in and type:servicebus but no topic tag
+        private static readonly string[] DefaultConsumeEdgeTags = ["direction:in", "type:servicebus"];
+
         /// <summary>
         /// OnMethodBegin callback
         /// </summary>
@@ -81,11 +84,12 @@ internal static CallTargetState OnMethodBegin<TTarget, TMessage>(TTarget instanc
 
                 var namespaceString = instance.Processor.EntityPath;
 
-                // TODO: we could pool these arrays to reduce allocations
                 // NOTE: the tags must be sorted in alphabetical order
                 var edgeTags = string.IsNullOrEmpty(namespaceString)
-                                    ? new[] { "direction:in", "type:servicebus" }
-                                    : new[] { "direction:in", $"topic:{namespaceString}", "type:servicebus" };
+                                    ? DefaultConsumeEdgeTags
+                                    : dataStreamsManager.GetOrCreateEdgeTags(
+                                        new ServiceBusEdgeTagCacheKey(namespaceString),
+                                        static k => ["direction:in", $"topic:{k.EntityPath}", "type:servicebus"]);
                 var msgSize = dataStreamsManager.IsInDefaultState ? 0 : AzureServiceBusCommon.GetMessageSize(message);
                 span.SetDataStreamsCheckpoint(
                     dataStreamsManager,
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/ServiceBus/ServiceBusEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/ServiceBus/ServiceBusEdgeTagCacheKey.cs
@@ -0,0 +1,33 @@
+// <copyright file="ServiceBusEdgeTagCacheKey.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>
+
+#nullable enable
+
+using System;
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.Azure.ServiceBus;
+
+/// <summary>
+/// Value-type cache key for Azure Service Bus consume edge tags. Using a named struct avoids boxing and
+/// is compatible with all supported target frameworks.
+/// </summary>
+internal readonly struct ServiceBusEdgeTagCacheKey : IEquatable<ServiceBusEdgeTagCacheKey>
+{
+    public readonly string EntityPath;
+
+    public ServiceBusEdgeTagCacheKey(string entityPath)
+    {
+        EntityPath = entityPath;
+    }
+
+    public bool Equals(ServiceBusEdgeTagCacheKey other)
+        => EntityPath == other.EntityPath;
+
+    public override bool Equals(object? obj)
+        => obj is ServiceBusEdgeTagCacheKey other && Equals(other);
+
+    public override int GetHashCode()
+        => EntityPath?.GetHashCode() ?? 0;
+}
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/IbmMq/GetIntegration.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/IbmMq/GetIntegration.cs
@@ -58,7 +58,9 @@ internal static CallTargetReturn OnMethodEnd<TTarget>(TTarget instance, Exceptio
                         }
 
                         var queueName = IbmMqHelper.SanitizeQueueName(instance.Name);
-                        var edgeTags = new[] { "direction:in", $"topic:{queueName}", $"type:{IbmMqConstants.QueueType}" };
+                        var edgeTags = dataStreams.GetOrCreateEdgeTags(
+                            new IbmMqEdgeTagCacheKey(queueName, isConsume: true),
+                            static k => ["direction:in", $"topic:{k.QueueName}", "type:ibmmq"]);
 
                         scope.Span.SetDataStreamsCheckpoint(
                             dataStreams,
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/IbmMq/IbmMqEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/IbmMq/IbmMqEdgeTagCacheKey.cs
@@ -0,0 +1,45 @@
+// <copyright file="IbmMqEdgeTagCacheKey.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>
+
+#nullable enable
+
+using System;
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.IbmMq;
+
+/// <summary>
+/// Value-type cache key for IBM MQ edge tags. Using a named struct avoids boxing and
+/// is compatible with all supported target frameworks.
+/// <see cref="IsConsume"/> distinguishes produce (direction:out) from consume (direction:in)
+/// so that both directions share a single cache type without key collision.
+/// </summary>
+internal readonly struct IbmMqEdgeTagCacheKey : IEquatable<IbmMqEdgeTagCacheKey>
+{
+    public readonly string QueueName;
+    public readonly bool IsConsume;
+
+    public IbmMqEdgeTagCacheKey(string queueName, bool isConsume)
+    {
+        QueueName = queueName;
+        IsConsume = isConsume;
+    }
+
+    public bool Equals(IbmMqEdgeTagCacheKey other)
+        => QueueName == other.QueueName && IsConsume == other.IsConsume;
+
+    public override bool Equals(object? obj)
+        => obj is IbmMqEdgeTagCacheKey other && Equals(other);
+
+    public override int GetHashCode()
+    {
+        unchecked
+        {
+            int hash = 17;
+            hash = (hash * 31) + (QueueName?.GetHashCode() ?? 0);
+            hash = (hash * 31) + IsConsume.GetHashCode();
+            return hash;
+        }
+    }
+}
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/IbmMq/PutIntegration.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/IbmMq/PutIntegration.cs
@@ -45,7 +45,9 @@ internal static CallTargetState OnMethodBegin<TTarget, TMessage, TOptions>(TTarg
                 if (dataStreams.IsEnabled && (instance).Instance != null && (msg).Instance != null)
                 {
                     var queueName = IbmMqHelper.SanitizeQueueName(instance.Name);
-                    var edgeTags = new[] { "direction:out", $"topic:{queueName}", $"type:{IbmMqConstants.QueueType}" };
+                    var edgeTags = dataStreams.GetOrCreateEdgeTags(
+                        new IbmMqEdgeTagCacheKey(queueName, isConsume: false),
+                        static k => ["direction:out", $"topic:{k.QueueName}", "type:ibmmq"]);
                     scope.Span.SetDataStreamsCheckpoint(dataStreams, CheckpointKind.Produce, edgeTags, msg.MessageLength, 0);
                     dataStreams.InjectPathwayContextAsBase64String(scope.Span.Context.PathwayContext, IbmMqHelper.GetHeadersAdapter(msg));
                 }
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Kafka/ConsumeEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Kafka/ConsumeEdgeTagCacheKey.cs
@@ -0,0 +1,46 @@
+// <copyright file="ConsumeEdgeTagCacheKey.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>
+
+#nullable enable
+
+using System;
+
+namespace Datadog.Trace.ClrProfiler.AutoInstrumentation.Kafka;
+
+/// <summary>
+/// Value-type cache key for consume edge tags. Using a named struct avoids boxing and
+/// is compatible with all supported target frameworks.
+/// </summary>
+internal readonly struct ConsumeEdgeTagCacheKey : IEquatable<ConsumeEdgeTagCacheKey>
+{
+    public readonly string GroupId;
+    public readonly string Topic;
+    public readonly string ClusterId;
+
+    public ConsumeEdgeTagCacheKey(string groupId, string topic, string clusterId)
+    {
+        GroupId = groupId;
+        Topic = topic;
+        ClusterId = clusterId;
+    }
+
+    public bool Equals(ConsumeEdgeTagCacheKey other)
+        => GroupId == other.GroupId && Topic == other.Topic && ClusterId == other.ClusterId;
+
+    public override bool Equals(object? obj)
+        => obj is ConsumeEdgeTagCacheKey other && Equals(other);
+
+    public override int GetHashCode()
+    {
+        unchecked
+        {
+            int hash = 17;
+            hash = (hash * 31) + (GroupId?.GetHashCode() ?? 0);
+            hash = (hash * 31) + (Topic?.GetHashCode() ?? 0);
+            hash = (hash * 31) + (ClusterId?.GetHashCode() ?? 0);
+            return hash;
+        }
+    }
+}
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Kafka/KafkaHelper.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Kafka/KafkaHelper.cs
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/RabbitMQ/RabbitMQConsumeEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/RabbitMQ/RabbitMQConsumeEdgeTagCacheKey.cs
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/RabbitMQ/RabbitMQIntegration.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/RabbitMQ/RabbitMQIntegration.cs
diff --git a/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/RabbitMQ/RabbitMQProduceEdgeTagCacheKey.cs b/tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/RabbitMQ/RabbitMQProduceEdgeTagCacheKey.cs
diff --git a/tracer/src/Datadog.Trace/DataStreamsMonitoring/EdgeTagCache.cs b/tracer/src/Datadog.Trace/DataStreamsMonitoring/EdgeTagCache.cs
diff --git a/tracer/test/Datadog.Trace.Tests/DataStreamsMonitoring/DataStreamsManagerTests.cs b/tracer/test/Datadog.Trace.Tests/DataStreamsMonitoring/DataStreamsManagerTests.cs

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,9 @@ internal static TResponse OnAsyncMethodEnd<TTarget, TResponse>(TTarget instance,`
`61`	`61`	`var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager;`
`62`	`62`	`if (dataStreamsManager is { IsEnabled: true })`
`63`	`63`	`{`
`64`		`- var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:kinesis" };`
	`64`	`+ var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(`
	`65`	`+ new KinesisEdgeTagCacheKey((string)state.State, isConsume: true),`
	`66`	`+ static k => ["direction:in", $"topic:{k.StreamName}", "type:kinesis"]);`
`65`	`67`	`foreach (var o in response.Records)`
`66`	`68`	`{`
`67`	`69`	`var record = o.DuckCast<IRecord>();`
Original file line number	Diff line number	Diff line change
`@@ -78,7 +78,9 @@ internal static CallTargetReturn<TResponse> OnMethodEnd<TTarget, TResponse>(TTar`
`78`	`78`	`var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager;`
`79`	`79`	`if (dataStreamsManager is { IsEnabled: true })`
`80`	`80`	`{`
`81`		`- var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:kinesis" };`
	`81`	`+ var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(`
	`82`	`+ new KinesisEdgeTagCacheKey((string)state.State, isConsume: true),`
	`83`	`+ static k => ["direction:in", $"topic:{k.StreamName}", "type:kinesis"]);`
`82`	`84`	`foreach (var o in response.Records)`
`83`	`85`	`{`
`84`	`86`	`var record = o.DuckCast<IRecord>();`
Original file line number	Diff line number	Diff line change
`@@ -37,8 +37,11 @@ public static CallTargetState BeforePublish<TPublishRequest>(TPublishRequest req`
`37`	`37`	`if (scope?.Span.Context is { } context && !string.IsNullOrEmpty(topicName))`
`38`	`38`	`{`
`39`	`39`	`var dataStreamsManager = tracer.TracerManager.DataStreamsManager;`
`40`		`- // avoid allocation if edgeTags are not going to be used`
`41`		`- var edgeTags = dataStreamsManager is { IsEnabled: true } ? ["direction:out", $"topic:{topicName}", "type:sns"] : Array.Empty<string>();`
	`40`	`+ var edgeTags = dataStreamsManager is { IsEnabled: true }`
	`41`	`+ ? dataStreamsManager.GetOrCreateEdgeTags(`
	`42`	`+ new SnsEdgeTagCacheKey(topicName!),`
	`43`	`+ static k => ["direction:out", $"topic:{k.TopicName}", "type:sns"])`
	`44`	`+ : Array.Empty<string>();`
`42`	`45`
`43`	`46`	`if (sendType == SendType.SingleMessage)`
`44`	`47`	`{`
Original file line number	Diff line number	Diff line change
`@@ -66,7 +66,9 @@ private static void InjectForSingleMessage<TSendMessageRequest>(Tracer tracer, T`
`66`	`66`	`var dataStreamsManager = tracer.TracerManager.DataStreamsManager;`
`67`	`67`	`if (dataStreamsManager != null && dataStreamsManager.IsEnabled)`
`68`	`68`	`{`
`69`		`- var edgeTags = new[] { "direction:out", $"topic:{queueName}", "type:sqs" };`
	`69`	`+ var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(`
	`70`	`+ new SqsEdgeTagCacheKey(queueName, isConsume: false),`
	`71`	`+ static k => ["direction:out", $"topic:{k.QueueName}", "type:sqs"]);`
`70`	`72`	`scope.Span.SetDataStreamsCheckpoint(dataStreamsManager, CheckpointKind.Produce, edgeTags, payloadSizeBytes: 0, timeInQueueMs: 0);`
`71`	`73`	`}`
`72`	`74`
`@@ -81,14 +83,18 @@ private static void InjectForBatch<TSendMessageBatchRequest>(Tracer tracer, TSen`
`81`	`83`	`return;`
`82`	`84`	`}`
`83`	`85`
`84`		`- var edgeTags = new[] { "direction:out", $"topic:{queueName}", "type:sqs" };`
	`86`	`+ var dataStreamsManager = tracer.TracerManager.DataStreamsManager;`
	`87`	`+ var edgeTags = dataStreamsManager is { IsEnabled: true }`
	`88`	`+ ? dataStreamsManager.GetOrCreateEdgeTags(`
	`89`	`+ new SqsEdgeTagCacheKey(queueName, isConsume: false),`
	`90`	`+ static k => ["direction:out", $"topic:{k.QueueName}", "type:sqs"])`
	`91`	`+ : new[] { "direction:out", $"topic:{queueName}", "type:sqs" };`
`85`	`92`	`foreach (var e in requestProxy.Entries)`
`86`	`93`	`{`
`87`	`94`	`var entry = e.DuckCast<IContainsMessageAttributes>();`
`88`	`95`	`if (entry != null)`
`89`	`96`	`{`
`90`	`97`	`// this has no effect if DSM is disabled`
`91`		`- var dataStreamsManager = tracer.TracerManager.DataStreamsManager;`
`92`	`98`	`scope.Span.SetDataStreamsCheckpoint(dataStreamsManager, CheckpointKind.Produce, edgeTags, payloadSizeBytes: 0, timeInQueueMs: 0);`
`93`	`99`	`// this needs to be done for context propagation even when DSM is disabled`
`94`	`100`	`// (when DSM is enabled, it injects the pathway context on top of the trace context)`
`@@ -148,7 +154,9 @@ internal static TResponse AfterReceive<TResponse>(TResponse response, Exception?`
`148`	`154`	`var dataStreamsManager = Tracer.Instance.TracerManager.DataStreamsManager;`
`149`	`155`	`if (dataStreamsManager is { IsEnabled: true })`
`150`	`156`	`{`
`151`		`- var edgeTags = new[] { "direction:in", $"topic:{(string)state.State}", "type:sqs" };`
	`157`	`+ var edgeTags = dataStreamsManager.GetOrCreateEdgeTags(`
	`158`	`+ new SqsEdgeTagCacheKey((string)state.State, isConsume: true),`
	`159`	`+ static k => ["direction:in", $"topic:{k.QueueName}", "type:sqs"]);`
`152`	`160`	`foreach (var o in response.Messages)`
`153`	`161`	`{`
`154`	`162`	`var message = o.DuckCast<IMessage>();`