DSM overhead optimizations (#8450)

# DSM Per-Message Overhead Optimizations

## Summary of changes

- **Edge-tag array caching**: Introduced `EdgeTagCache<TKey>` and
`BacklogTagCache<TKey>` — process-wide, per-type `ConcurrentDictionary`
caches that intern edge-tag arrays and backlog-tag strings so they are
only allocated once per unique key (topic/group/cluster combination).
- **Node-hash caching**: Added a `NodeHashCacheEntry`/`NodeHashSnapshot`
mechanism inside `DataStreamsManager` that memoizes the expensive
`CalculateNodeHash` result per `(edgeTags[], nodeHashBase)` pair. Reads
are lock-free via a volatile field; writes acquire a per-entry lock only
on cache miss or base change.
- **Zero-allocation context encode/decode (net core 3.1+)**: Added
`PathwayContextEncoder.EncodeInto` and a `Span<byte>`-based `Decode`
overload; `DataStreamsContextPropagator` uses `stackalloc` buffers on
.NET Core 3.1+ to avoid intermediate `byte[]` heap allocations on every
produce/consume.
- **Reference-equality dictionary comparers**: `DataStreamsAggregator`
and `DataStreamsManager._nodeHashCache` now use reference-equality
comparers backed by `RuntimeHelpers.GetHashCode`, which is safe because
all keys are interned by the caches above.
- **Drain-signal instead of sleep**: Replaced the 10 ms `Thread.Sleep`
polling loop in `DataStreamsWriter` with a `ManualResetEventSlim` that
wakes immediately when the queue reaches 1 000 items or after a 500 ms
timeout, eliminating unnecessary context switches.
- **Integration-specific cache-key structs**: Added `readonly struct`
cache keys (`ConsumeEdgeTagCacheKey`, `ProduceEdgeTagCacheKey`,
`CommitBacklogTagCacheKey`, `ProduceBacklogTagCacheKey`) for Kafka;
equivalent structs for AWS SQS/SNS/Kinesis, Azure Service Bus, IBM MQ,
and RabbitMQ.
- **Minor hot-path fix (Kafka)**: The
`Remove(TemporaryBase64PathwayContext)` header scan is now skipped when
`KafkaCreateConsumerScopeEnabled=true` (the default), avoiding an O(n)
scan on every message.
- **`LastConsumePathway` guard removed**: Dropped the redundant `!=
null` guard on the produce path that required an `AsyncLocal` read
before the actual `AsyncLocal` read.

## Reason for change

DSM instrumentation runs on the hot path of every instrumented message.
Profiling revealed that the dominant allocations were:
1. A new `string[]` edge-tag array on every produce/consume call.
2. A `CalculateNodeHash` call (hashing over all edge tags) on every
checkpoint.
3. Intermediate `byte[]` arrays for pathway context Base64
encoding/decoding.
4. Unnecessary CPU spin from a fixed 10 ms sleep between drain cycles.

These optimizations target p99 and throughput benchmarks for Kafka, SQS,
SNS, RabbitMQ, IBM MQ, Azure Service Bus, and Kinesis instrumentation.

## Implementation details

### Caching strategy

`EdgeTagCache<TKey>` and `BacklogTagCache<TKey>` use the
static-generic-class pattern (`static class Foo<T>` with a static field)
to give each integration its own dictionary instance without any runtime
dispatch. The key type is a `readonly struct` implementing
`IEquatable<TKey>`, which prevents boxing in `ConcurrentDictionary`
lookups.

The caches are bounded at `MaxEdgeTagCacheSize = 1000` entries. Once
that limit is reached, new keys are computed on the fly (no caching) to
prevent unbounded memory growth from high-cardinality identifiers.

### Node-hash caching

`_nodeHashCache` is keyed by `string[]` **identity** (not value
equality) because the arrays themselves are interned by
`EdgeTagCache<TKey>`. Each entry holds a volatile `NodeHashSnapshot`
(`nodeHashBase` + `NodeHash`). On every checkpoint:

1. Look up the array reference — O(1) identity hash.
2. Read the volatile snapshot — lock-free.
3. If the base matches, return immediately.
4. Otherwise, acquire the per-entry lock, double-check, compute, and
publish a new snapshot.

### Zero-allocation encode/decode

`PathwayContextEncoder.EncodeInto(PathwayContext, Span<byte>)` writes
directly into a caller-supplied buffer. `DataStreamsContextPropagator`
stackallocs `MaxEncodedSize` (26 bytes) and `MaxBase64EncodedSize` (36
bytes) on the stack and uses `Base64.EncodeToUtf8`/`DecodeFromUtf8`
in-place. The only unavoidable allocation is the final `ToArray()`
passed to `headers.Add`, because Kafka takes ownership of the byte
array.

This path is guarded by `#if NETCOREAPP3_1_OR_GREATER`; .NET Framework
falls back to the original heap-allocating path.

### Drain signal

`DataStreamsWriter` previously slept 10 ms unconditionally between drain
iterations, burning CPU and adding ~10 ms latency per batch even under
load. The new `ManualResetEventSlim` is signalled immediately when
either queue exceeds `DrainThreshold` (1 000 items), capping worst-case
latency at `DrainTimeoutMs` (500 ms) while eliminating idle wakeups.

## Test coverage

- `DataStreamsManagerTests`: new unit tests verify that
`GetOrCreateEdgeTags` and `GetOrCreateBacklogTags` return the **same
array/string reference** on repeated calls with the same key, and
distinct references for different keys. Tests cover Kafka
produce/consume, RabbitMQ produce/consume, and generic key types.
- `PathwayContextEncoderTests`: existing encode/decode round-trip tests
pass against the new `Span<byte>` overloads.
- All existing DSM tests continue to pass.

## Other details

- The `MaxEdgeTagCacheSize` constant is `internal` to allow unit tests
to verify the overflow/bypass behavior.
- No public API surface changes; all new types are `internal`.
- `.NET Framework` code paths are unchanged — all `Span`-based
optimizations are gated behind `#if NETCOREAPP3_1_OR_GREATER`.

Author: kr-igor

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/ContextPropagation.cs

@@ -58,13 +58,15 @@ public static void InjectTraceIntoData<TRecordRequest>(Tracer tracer, TRecordReq
             }
 
             var propagatedContext = new Dictionary<string, object>();
-            if (scope.Span.Context != null && !string.IsNullOrEmpty(streamName))
+            if (scope.Span.Context != null && !StringUtil.IsNullOrEmpty(streamName))
             {
                 var dataStreamsManager = tracer.TracerManager.DataStreamsManager;
-                if (dataStreamsManager != null && dataStreamsManager.IsEnabled)
+                if (dataStreamsManager is { IsEnabled: true })
                 {
                     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,

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>();

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>();

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/Kinesis/KinesisEdgeTagCacheKey.cs

@@ -0,0 +1,16 @@
+// <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
+
+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 record struct KinesisEdgeTagCacheKey(string StreamName, bool IsConsume);

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SNS/AwsSnsHandlerCommon.cs

@@ -34,11 +34,14 @@ public static CallTargetState BeforePublish<TPublishRequest>(TPublishRequest req
             tags.TopicName = topicName;
         }
 
-        if (scope?.Span.Context is { } context && !string.IsNullOrEmpty(topicName))
+        if (scope?.Span.Context is { } context && !StringUtil.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"])
+                               : [];
 
             if (sendType == SendType.SingleMessage)
             {

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SNS/SnsEdgeTagCacheKey.cs

@@ -0,0 +1,14 @@
+// <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
+
+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 record struct SnsEdgeTagCacheKey(string TopicName);

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"])
+                           : [];
         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>();

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/AWS/SQS/SqsEdgeTagCacheKey.cs

@@ -0,0 +1,16 @@
+// <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
+
+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 record struct SqsEdgeTagCacheKey(string QueueName, bool IsConsume);

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,

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Azure/ServiceBus/ServiceBusEdgeTagCacheKey.cs

@@ -0,0 +1,14 @@
+// <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
+
+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 record struct ServiceBusEdgeTagCacheKey(string EntityPath);