From 82506534e94c2a5a89818d62993850abbd7f6cc0 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 24 Jul 2026 12:27:05 -0400 Subject: [PATCH 1/6] 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:, 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 --- .../common/metrics/AdditionalTagsSchema.java | 3 + .../trace/common/metrics/Aggregator.java | 6 ++ .../trace/common/metrics/CoreHandlers.java | 5 +- .../trace/common/metrics/PeerTagSchema.java | 2 + .../metrics/AdditionalTagsSchemaTest.java | 29 ++++++ .../trace/api/metrics/StatsMetrics.java | 92 +++++++++++++++++++ .../api/telemetry/CoreMetricCollector.java | 18 ++++ .../trace/api/metrics/StatsMetricsTest.java | 89 ++++++++++++++++++ 8 files changed, 243 insertions(+), 1 deletion(-) create mode 100644 internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java create mode 100644 internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AdditionalTagsSchema.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AdditionalTagsSchema.java index a1a95da2f4e..1965f06d3ba 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AdditionalTagsSchema.java +++ b/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); } } } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java index 34d7b2a83fe..c2400945595 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java +++ b/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 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); } } } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/CoreHandlers.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/CoreHandlers.java index 9214e34a89b..fc6223157ca 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/CoreHandlers.java +++ b/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); } } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/PeerTagSchema.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/PeerTagSchema.java index 98360d44d47..28d317daf35 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/PeerTagSchema.java +++ b/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); } } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AdditionalTagsSchemaTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AdditionalTagsSchemaTest.java index 475cf86d139..65ebf357055 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AdditionalTagsSchemaTest.java +++ b/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 diff --git a/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java b/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java new file mode 100644 index 00000000000..5a6c49e2602 --- /dev/null +++ b/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. + * + *

Each collapse "reason" (e.g. {@code collapsed:additional_metric_tags}, {@code + * oversized:additional_metric_tags}, {@code collapsed:peer_tags}, {@code collapsed:}, {@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. + * + *

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 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 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; + } + } +} diff --git a/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java b/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java index d33fcf1529d..f4ee4ba1104 100644 --- a/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java +++ b/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 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 diff --git a/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java b/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java new file mode 100644 index 00000000000..25dacb93da1 --- /dev/null +++ b/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 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 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()); + } +} From 56a08367b2f029fae49d7073a40e61bc37416195 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 27 Jul 2026 11:21:32 -0400 Subject: [PATCH 2/6] Hoist stats whole_key collapse counter off the per-dropped-span map lookup The collapsed:whole_key counter is incremented once per dropped span on the aggregator thread (Aggregator, findOrInsert == null branch), unlike every other collapse reason which is batched once per reporting cycle. Under a cardinality explosion the aggregate table pins at cap and every arriving span is dropped, turning that path hot on an already-bottlenecked thread -- yet it went through a ConcurrentHashMap.computeIfAbsent lookup on StatsMetrics per span. Pre-create the whole_key TaggedCounter at construction (still registered in the reason map so the telemetry drain picks it up unchanged) and add a dedicated onWholeKeyCollapse() that increments the cached reference directly. The hot path no longer touches the map; the batched reasons still route through onCollapsedSpans as before. Co-Authored-By: Claude Opus 4.8 --- .../trace/common/metrics/Aggregator.java | 6 +----- .../trace/api/metrics/StatsMetrics.java | 19 +++++++++++++++++++ 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java index c2400945595..27f81943555 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java @@ -16,10 +16,6 @@ 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 inbox; @@ -161,7 +157,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); + StatsMetrics.getInstance().onWholeKeyCollapse(); } } } diff --git a/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java b/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java index 5a6c49e2602..8e49eb09b19 100644 --- a/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java +++ b/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java @@ -24,6 +24,7 @@ */ 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(); @@ -31,6 +32,15 @@ public final class StatsMetrics { // that reason; the reason set is bounded, so this never grows unboundedly. private final ConcurrentMap 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; } @@ -51,6 +61,15 @@ public void onCollapsedSpans(String reason, long count) { .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 getTaggedCounters() { return this.collapsedByReason.values(); } From 01daa8b292b448c4b8745f1a5b26c339da70175e Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 27 Jul 2026 11:21:44 -0400 Subject: [PATCH 3/6] Test whole_key collapse increments the pre-created StatsMetrics counter Co-Authored-By: Claude Opus 4.8 --- .../trace/api/metrics/StatsMetricsTest.java | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java b/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java index 25dacb93da1..e2b52f84a2f 100644 --- a/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java +++ b/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java @@ -73,6 +73,28 @@ void nonPositiveCountsAreIgnored() { 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(); From 6431932502ad9661eac78d6a43560f513925533a Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Mon, 27 Jul 2026 11:55:54 -0400 Subject: [PATCH 4/6] Guard stats telemetry drain against delta loss when the queue is full getValueAndReset() resets a counter's delta baseline, so calling it before the offer() succeeds means a full metricsQueue silently drops that delta forever -- exactly when it matters most (telemetry backing up because the agent is unreachable). Check remainingCapacity() before reading any counter and stop the drain early instead; the untouched counters are collected on the next cycle. Co-Authored-By: Claude Opus 4.8 --- .../datadog/trace/api/telemetry/CoreMetricCollector.java | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java b/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java index f4ee4ba1104..40df41ff1b5 100644 --- a/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java +++ b/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java @@ -70,6 +70,12 @@ public void prepareMetrics() { // Collect client-side trace-stats span-collapse metrics, tagged by collapse reason. 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 From c0ce934d3b87432daba8398c4d528cafd74cf1b0 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 28 Jul 2026 12:27:46 -0400 Subject: [PATCH 5/6] Collect bounded stats collapse counters before the unbounded span registry The span-metric registry holds one entry per instrumentation name and can exceed the 1024-entry telemetry queue on its own. Draining it before the small, fixed set of client-side stats collapse counters could fill the queue and starve those counters indefinitely. Collect them first so they are always emitted; the pre-read capacity guard still protects their delta baseline. Co-Authored-By: Claude Opus 4.8 --- .../api/telemetry/CoreMetricCollector.java | 48 ++++++++++--------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java b/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java index 40df41ff1b5..27cf17b7201 100644 --- a/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java +++ b/internal-api/src/main/java/datadog/trace/api/telemetry/CoreMetricCollector.java @@ -33,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(); @@ -67,28 +93,6 @@ public void prepareMetrics() { break; } } - - // Collect client-side trace-stats span-collapse metrics, tagged by collapse reason. - 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; - } - } } @Override From 869b8b724253ac5690d1c6e15c1412186ab87b22 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 28 Jul 2026 12:27:48 -0400 Subject: [PATCH 6/6] Drain StatsMetrics singleton deltas after each test StatsMetrics is a process-wide singleton; these tests left per-reason deltas behind, leaking stats.collapsed_spans metrics into CoreMetricCollectorTest when they shared a Gradle worker and making its exact span-metric count assertion order-dependent. Reset every counter in @AfterEach. Co-Authored-By: Claude Opus 4.8 --- .../datadog/trace/api/metrics/StatsMetricsTest.java | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java b/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java index e2b52f84a2f..abd1b782843 100644 --- a/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java +++ b/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java @@ -7,6 +7,7 @@ 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; /** @@ -20,6 +21,16 @@ private static Map countersByTag() { .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();