DataDog · dougqh · Jul 24, 2026 · Jul 27, 2026 · Jul 27, 2026 · Jul 27, 2026
@@ -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);
     }
   }
 }
@@ -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;
@@ -156,6 +157,7 @@ public void accept(InboxItem item) {
         } else {
           // table at cap with no stale entry available to evict
           healthMetrics.onStatsAggregateDropped();
+          StatsMetrics.getInstance().onWholeKeyCollapse();
         }
       }
     }

@@ -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);
       }
     }

@@ -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);
     }
   }
 

@@ -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

@@ -0,0 +1,111 @@
+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";
+  static final String COLLAPSED_WHOLE_KEY = "collapsed:whole_key";
+
+  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<>();
+
+  // The one reason counted per dropped span on the hot aggregator path (aggregate table at cap);
+  // every other reason is batched once per reporting cycle. Pre-created and cached so the per-span
+  // increment is a direct counter hit rather than a map lookup -- this matters precisely when a
+  // cardinality explosion pins the table at cap and every arriving span is dropped, turning a cold
+  // path hot. Still registered in the map above, so the telemetry drain sees it with the rest.
+  private final TaggedCounter wholeKeyCollapses =
+      this.collapsedByReason.computeIfAbsent(
+          COLLAPSED_WHOLE_KEY, tag -> new TaggedCounter(COLLAPSED_SPANS, tag));
+
+  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);
+  }
+
+  /**
+   * Records a single whole-key collapse: a span dropped because the aggregate table was at cap with
+   * no entry to evict. Increments the pre-created {@link #COLLAPSED_WHOLE_KEY} counter directly,
+   * keeping the per-dropped-span aggregator path off the reason map.
+   */
+  public void onWholeKeyCollapse() {
+    this.wholeKeyCollapses.counter.addAndGet(1);
+  }
+
+  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;
+    }
+  }
+}
@@ -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;
 
@@ -31,6 +33,32 @@ private CoreMetricCollector() {
 
   @Override
   public void prepareMetrics() {
+    // Collect the bounded, high-value client-side trace-stats span-collapse counters first, tagged
+    // by collapse reason. There is only a small, fixed set of these; the span-metric registry below
+    // is unbounded (one entry per instrumentation name), so draining it first could fill the queue
+    // and starve the collapse counters indefinitely under high instrumentation counts. Collecting
+    // them up front guarantees they are emitted.
+    for (StatsMetrics.TaggedCounter counter : this.statsMetrics.getTaggedCounters()) {
+      if (this.metricsQueue.remainingCapacity() == 0) {
+        // Queue full: stop before reading any more counters. getValueAndReset() below resets the
+        // counter's delta baseline, so resetting one we then fail to enqueue would drop that delta
+        // for good; the untouched counters are picked up on the next collection cycle.
+        break;
+      }
+      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;
+      }
+    }
+
     // Collect span metrics
     for (SpanMetricsImpl spanMetrics : this.spanMetricRegistry.getSpanMetrics()) {
       String tag = INTEGRATION_NAME_TAG + spanMetrics.getInstrumentationName();

@@ -0,0 +1,122 @@
+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.AfterEach;
+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()));
+  }
+
+  @AfterEach
+  void drainDeltas() {
+    // StatsMetrics is a process-wide singleton; these tests leave per-reason deltas behind. Drain
+    // every counter so they do not leak into another collector's expectations when tests share a
+    // Gradle worker (e.g. CoreMetricCollectorTest asserting an exact span-metric count).
+    for (StatsMetrics.TaggedCounter counter : StatsMetrics.getInstance().getTaggedCounters()) {
+      counter.getValueAndReset();
+    }
+  }
+
+  @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 wholeKeyCollapseIncrementsPreCreatedCounter() {
+    StatsMetrics metrics = StatsMetrics.getInstance();
+
+    // The whole_key counter is pre-created at construction, so it is always present in the drain.
+    StatsMetrics.TaggedCounter counter = countersByTag().get(StatsMetrics.COLLAPSED_WHOLE_KEY);
+    assertEquals(StatsMetrics.COLLAPSED_SPANS, counter.getName());
+    assertEquals(StatsMetrics.COLLAPSED_WHOLE_KEY, counter.getTag());
+
+    long before = counter.getValue();
+    metrics.onWholeKeyCollapse();
+    metrics.onWholeKeyCollapse();
+    assertEquals(before + 2, counter.getValue(), "each call increments the counter by one");
+
+    // Routing the same tag through onCollapsedSpans hits the same pre-created counter instance.
+    assertTrue(
+        countersByTag().get(StatsMetrics.COLLAPSED_WHOLE_KEY) == counter,
+        "whole_key resolves to a single stable counter");
+    metrics.onCollapsedSpans(StatsMetrics.COLLAPSED_WHOLE_KEY, 3);
+    assertEquals(before + 5, counter.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());
+  }
+}