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..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 @@ -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(); } } } 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..8e49eb09b19 --- /dev/null +++ b/internal-api/src/main/java/datadog/trace/api/metrics/StatsMetrics.java @@ -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. + * + *

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"; + 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 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 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..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 @@ -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; @@ -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(); 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..abd1b782843 --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/api/metrics/StatsMetricsTest.java @@ -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 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 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()); + } +}