Report stats span collapses over telemetry (stats.collapsed_spans)

The client-side trace-stats engine already emits a statsd metric
(datadog.tracer.stats.collapsed_spans) whenever spans collapse under a
cardinality/length limit or a whole-key table-drop. That signal is invisible
when no dogstatsd sink is configured, since the aggregator's HealthMetrics is
NO_OP in that case.

Add a parallel telemetry counter, stats.collapsed_spans (namespace tracers),
tagged by collapse reason (collapsed:<field>, collapsed:peer_tags,
collapsed:additional_metric_tags, oversized:additional_metric_tags,
collapsed:whole_key). It is fed directly at each reset/drop site alongside the
existing HealthMetrics call, so it is independent of the statsd sink, and
drained by CoreMetricCollector like the baggage counters. The reason set is
bounded by construction, so the dynamic per-reason map cannot itself grow
unboundedly.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: dougqh

dd-trace-core/src/main/java/datadog/trace/common/metrics/AdditionalTagsSchema.java

@@ -1,5 +1,6 @@
 package datadog.trace.common.metrics;
 
+import datadog.trace.api.metrics.StatsMetrics;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.core.monitor.HealthMetrics;
 import java.util.ArrayList;
@@ -124,9 +125,11 @@ void resetHandlers(HealthMetrics healthMetrics, CardinalityLimitReporter reporte
     // the approved Cardinality Limits RFC.
     if (totalCollapsed > 0) {
       healthMetrics.onTagCardinalityBlocked(COLLAPSED_STATSD_TAG, totalCollapsed);
+      StatsMetrics.getInstance().onCollapsedSpans(COLLAPSED_STATSD_TAG[0], totalCollapsed);
     }
     if (totalOversized > 0) {
       healthMetrics.onTagCardinalityBlocked(OVERSIZED_STATSD_TAG, totalOversized);
+      StatsMetrics.getInstance().onCollapsedSpans(OVERSIZED_STATSD_TAG[0], totalOversized);
     }
   }
 }

dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java

@@ -2,6 +2,7 @@
 
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 
+import datadog.trace.api.metrics.StatsMetrics;
 import datadog.trace.common.metrics.SignalItem.ClearSignal;
 import datadog.trace.common.metrics.SignalItem.StopSignal;
 import datadog.trace.core.monitor.HealthMetrics;
@@ -15,6 +16,10 @@ final class Aggregator implements Runnable {
 
   private static final long DEFAULT_SLEEP_MILLIS = 10;
 
+  // Telemetry collapse reason for a whole-key drop (aggregate table at cap, no evictable entry);
+  // mirrors the statsd "collapsed:whole_key" tag on datadog.tracer.stats.collapsed_spans.
+  private static final String COLLAPSED_WHOLE_KEY_TAG = "collapsed:whole_key";
+
   private static final Logger log = LoggerFactory.getLogger(Aggregator.class);
 
   private final MessagePassingQueue<InboxItem> inbox;
@@ -156,6 +161,7 @@ public void accept(InboxItem item) {
         } else {
           // table at cap with no stale entry available to evict
           healthMetrics.onStatsAggregateDropped();
+          StatsMetrics.getInstance().onCollapsedSpans(COLLAPSED_WHOLE_KEY_TAG, 1);
         }
       }
     }

dd-trace-core/src/main/java/datadog/trace/common/metrics/CoreHandlers.java

@@ -1,6 +1,7 @@
 package datadog.trace.common.metrics;
 
 import datadog.trace.api.Config;
+import datadog.trace.api.metrics.StatsMetrics;
 import datadog.trace.core.monitor.HealthMetrics;
 
 /**
@@ -92,7 +93,9 @@ void reset(HealthMetrics healthMetrics, CardinalityLimitReporter reporter) {
     for (PropertyCardinalityHandler h : handlers) {
       long numBlocked = h.reset();
       if (numBlocked > 0) {
-        healthMetrics.onTagCardinalityBlocked(h.statsDTag(), numBlocked);
+        String[] statsDTag = h.statsDTag();
+        healthMetrics.onTagCardinalityBlocked(statsDTag, numBlocked);
+        StatsMetrics.getInstance().onCollapsedSpans(statsDTag[0], numBlocked);
         reporter.record(h.name, numBlocked);
       }
     }

dd-trace-core/src/main/java/datadog/trace/common/metrics/PeerTagSchema.java

@@ -4,6 +4,7 @@
 
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.trace.api.Config;
+import datadog.trace.api.metrics.StatsMetrics;
 import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
 import datadog.trace.core.monitor.HealthMetrics;
 import java.util.Set;
@@ -143,6 +144,7 @@ void resetHandlers(HealthMetrics healthMetrics, CardinalityLimitReporter reporte
     // approved Cardinality Limits RFC.
     if (totalCollapsed > 0) {
       healthMetrics.onTagCardinalityBlocked(COLLAPSED_STATSD_TAG, totalCollapsed);
+      StatsMetrics.getInstance().onCollapsedSpans(COLLAPSED_STATSD_TAG[0], totalCollapsed);
     }
   }
 

dd-trace-core/src/test/java/datadog/trace/common/metrics/AdditionalTagsSchemaTest.java

@@ -8,6 +8,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.verifyNoMoreInteractions;
 
+import datadog.trace.api.metrics.StatsMetrics;
 import datadog.trace.core.monitor.HealthMetrics;
 import java.util.Arrays;
 import java.util.Collections;
@@ -104,6 +105,34 @@ void resetHandlersReportsCardinalityCollapseUnderFieldName() {
     verifyNoMoreInteractions(metrics);
   }
 
+  @Test
+  void resetHandlersFeedsCollapseCountToTelemetry() {
+    // The reset site also feeds the telemetry counter (independent of the statsd HealthMetrics
+    // sink) under the same "collapsed:additional_metric_tags" reason tag.
+    AdditionalTagsSchema schema =
+        AdditionalTagsSchema.from(Collections.singleton("region"), 1, true);
+    int region = indexOf(schema, "region");
+    schema.register(region, "us-east-1"); // within budget
+    schema.register(region, "eu-west-1"); // collapsed (cardinality)
+    schema.register(region, "ap-south-1"); // collapsed (cardinality)
+
+    // Drain any pre-existing delta so the assertion measures only this reset's contribution.
+    drainTelemetryDelta("collapsed:additional_metric_tags");
+    schema.resetHandlers(mock(HealthMetrics.class), new CardinalityLimitReporter());
+
+    assertEquals(2L, drainTelemetryDelta("collapsed:additional_metric_tags"));
+  }
+
+  /** Reads and resets the telemetry delta for {@code reason}; 0 if no counter exists yet. */
+  private static long drainTelemetryDelta(String reason) {
+    for (StatsMetrics.TaggedCounter counter : StatsMetrics.getInstance().getTaggedCounters()) {
+      if (reason.equals(counter.getTag())) {
+        return counter.getValueAndReset();
+      }
+    }
+    return 0L;
+  }
+
   @Test
   void resetHandlersAggregatesCardinalityCollapseAcrossKeysUnderOneFieldTag() {
     // Two keys each collapse: the field-level health metric sums both under a single

internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java

@@ -0,0 +1,92 @@
+package datadog.trace.api.metrics;
+
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Telemetry counters for client-side trace-stats span collapses. Mirrors the statsd {@code
+ * datadog.tracer.stats.collapsed_spans} metric so the same signal is visible over telemetry, which
+ * (unlike statsd) is always wired regardless of whether a dogstatsd sink is configured -- the stats
+ * aggregator's {@code HealthMetrics} is {@code NO_OP} when health metrics are disabled.
+ *
+ * <p>Each collapse "reason" (e.g. {@code collapsed:additional_metric_tags}, {@code
+ * oversized:additional_metric_tags}, {@code collapsed:peer_tags}, {@code collapsed:<field>}, {@code
+ * collapsed:whole_key}) is a distinct telemetry tag on a single {@code stats.collapsed_spans}
+ * counter. The reason set is bounded and low-cardinality by construction (the cardinality limits
+ * themselves guarantee it), so a dynamic map keyed by reason cannot itself blow up.
+ *
+ * <p>Counters are incremented from the single stats-aggregator thread and drained from the
+ * telemetry thread: the backing map is a {@link ConcurrentMap} and each counter is an {@link
+ * AtomicLong}, so neither side needs external synchronization. {@link
+ * TaggedCounter#getValueAndReset()} is only called from the draining thread.
+ */
+public final class StatsMetrics {
+  static final String COLLAPSED_SPANS = "stats.collapsed_spans";
+
+  private static final StatsMetrics INSTANCE = new StatsMetrics();
+
+  // reason tag (e.g. "collapsed:additional_metric_tags") -> counter. Created on first collapse for
+  // that reason; the reason set is bounded, so this never grows unboundedly.
+  private final ConcurrentMap<String, TaggedCounter> collapsedByReason = new ConcurrentHashMap<>();
+
+  public static StatsMetrics getInstance() {
+    return INSTANCE;
+  }
+
+  private StatsMetrics() {}
+
+  /**
+   * Records {@code count} spans collapsed under the given {@code reason} tag (e.g. {@code
+   * collapsed:additional_metric_tags}). No-op for a non-positive count.
+   */
+  public void onCollapsedSpans(String reason, long count) {
+    if (count <= 0) {
+      return;
+    }
+    collapsedByReason
+        .computeIfAbsent(reason, tag -> new TaggedCounter(COLLAPSED_SPANS, tag))
+        .counter
+        .addAndGet(count);
+  }
+
+  public Collection<TaggedCounter> getTaggedCounters() {
+    return this.collapsedByReason.values();
+  }
+
+  /** A named, single-tag counter drained as a telemetry {@code count} metric. */
+  public static final class TaggedCounter implements CoreCounter {
+    private final String name;
+    private final String tag;
+    private final AtomicLong counter = new AtomicLong();
+    private long previousCount;
+
+    TaggedCounter(String name, String tag) {
+      this.name = name;
+      this.tag = tag;
+    }
+
+    @Override
+    public String getName() {
+      return this.name;
+    }
+
+    public String getTag() {
+      return this.tag;
+    }
+
+    @Override
+    public long getValue() {
+      return this.counter.get();
+    }
+
+    @Override
+    public long getValueAndReset() {
+      long count = this.counter.get();
+      long delta = count - this.previousCount;
+      this.previousCount = count;
+      return delta;
+    }
+  }
+}

internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java

@@ -4,6 +4,7 @@
 import datadog.trace.api.metrics.CoreCounter;
 import datadog.trace.api.metrics.SpanMetricRegistryImpl;
 import datadog.trace.api.metrics.SpanMetricsImpl;
+import datadog.trace.api.metrics.StatsMetrics;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -18,6 +19,7 @@ public class CoreMetricCollector implements MetricCollector<CoreMetricCollector.
   private static final CoreMetricCollector INSTANCE = new CoreMetricCollector();
   private final SpanMetricRegistryImpl spanMetricRegistry = SpanMetricRegistryImpl.getInstance();
   private final BaggageMetrics baggageMetrics = BaggageMetrics.getInstance();
+  private final StatsMetrics statsMetrics = StatsMetrics.getInstance();
 
   private final BlockingQueue<CoreMetric> metricsQueue;
 
@@ -65,6 +67,22 @@ public void prepareMetrics() {
         break;
       }
     }
+
+    // Collect client-side trace-stats span-collapse metrics, tagged by collapse reason.
+    for (StatsMetrics.TaggedCounter counter : this.statsMetrics.getTaggedCounters()) {
+      long value = counter.getValueAndReset();
+      if (value == 0) {
+        // Skip not updated counters
+        continue;
+      }
+      CoreMetric metric =
+          new CoreMetric(
+              METRIC_NAMESPACE, true, counter.getName(), "count", value, counter.getTag());
+      if (!this.metricsQueue.offer(metric)) {
+        // Stop adding metrics if the queue is full
+        break;
+      }
+    }
   }
 
   @Override

internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java

@@ -0,0 +1,89 @@
+package datadog.trace.api.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link StatsMetrics}. The instance is a process-wide singleton, so each test uses
+ * its own uniquely-named collapse reasons to stay isolated from other tests' counters.
+ */
+class StatsMetricsTest {
+
+  private static Map<String, StatsMetrics.TaggedCounter> countersByTag() {
+    return StatsMetrics.getInstance().getTaggedCounters().stream()
+        .collect(Collectors.toMap(StatsMetrics.TaggedCounter::getTag, Function.identity()));
+  }
+
+  @Test
+  void accumulatesPerReasonAndEmitsResetDeltas() {
+    StatsMetrics metrics = StatsMetrics.getInstance();
+    String reason = "collapsed:test_accumulate";
+
+    metrics.onCollapsedSpans(reason, 3);
+    metrics.onCollapsedSpans(reason, 4);
+
+    StatsMetrics.TaggedCounter counter = countersByTag().get(reason);
+    assertEquals(StatsMetrics.COLLAPSED_SPANS, counter.getName());
+    assertEquals(reason, counter.getTag());
+    assertEquals(7, counter.getValue(), "getValue reports the running total");
+
+    // First drain returns the whole accumulated delta; a second drain with no activity returns 0.
+    assertEquals(7, counter.getValueAndReset(), "first drain returns the accumulated delta");
+    assertEquals(0, counter.getValueAndReset(), "no new activity -> zero delta");
+
+    metrics.onCollapsedSpans(reason, 5);
+    assertEquals(5, counter.getValueAndReset(), "only the post-drain increment is returned");
+  }
+
+  @Test
+  void separateReasonsGetSeparateCounters() {
+    StatsMetrics metrics = StatsMetrics.getInstance();
+    String collapsed = "collapsed:test_separate";
+    String oversized = "oversized:test_separate";
+
+    metrics.onCollapsedSpans(collapsed, 2);
+    metrics.onCollapsedSpans(oversized, 9);
+
+    Map<String, StatsMetrics.TaggedCounter> counters = countersByTag();
+    assertEquals(2, counters.get(collapsed).getValue());
+    assertEquals(9, counters.get(oversized).getValue());
+  }
+
+  @Test
+  void nonPositiveCountsAreIgnored() {
+    StatsMetrics metrics = StatsMetrics.getInstance();
+    String reason = "collapsed:test_nonpositive";
+
+    metrics.onCollapsedSpans(reason, 0);
+    metrics.onCollapsedSpans(reason, -5);
+
+    // No counter is created for a reason that never saw a positive count.
+    assertNull(countersByTag().get(reason), "no counter created for non-positive counts");
+
+    metrics.onCollapsedSpans(reason, 4);
+    assertEquals(4, countersByTag().get(reason).getValue());
+    // A later non-positive count leaves the running total untouched.
+    metrics.onCollapsedSpans(reason, -1);
+    assertEquals(4, countersByTag().get(reason).getValue());
+  }
+
+  @Test
+  void reasonCounterIsStableAcrossLookups() {
+    StatsMetrics metrics = StatsMetrics.getInstance();
+    String reason = "collapsed:test_stable";
+
+    metrics.onCollapsedSpans(reason, 1);
+    StatsMetrics.TaggedCounter first = countersByTag().get(reason);
+    metrics.onCollapsedSpans(reason, 1);
+    StatsMetrics.TaggedCounter second = countersByTag().get(reason);
+
+    assertTrue(first == second, "same reason maps to the same counter instance");
+    assertEquals(2, first.getValue());
+  }
+}