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..474650c8977 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 @@ -35,6 +35,12 @@ public void prepareMetrics() { for (SpanMetricsImpl spanMetrics : this.spanMetricRegistry.getSpanMetrics()) { String tag = INTEGRATION_NAME_TAG + spanMetrics.getInstrumentationName(); for (CoreCounter counter : spanMetrics.getCounters()) { + 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 @@ -51,6 +57,12 @@ public void prepareMetrics() { // Collect baggage metrics for (BaggageMetrics.TaggedCounter counter : this.baggageMetrics.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 diff --git a/internal-api/src/test/java/datadog/trace/api/telemetry/CoreMetricCollectorQueueFullTest.java b/internal-api/src/test/java/datadog/trace/api/telemetry/CoreMetricCollectorQueueFullTest.java new file mode 100644 index 00000000000..e66bf71090b --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/CoreMetricCollectorQueueFullTest.java @@ -0,0 +1,91 @@ +package datadog.trace.api.telemetry; + +import static datadog.trace.api.telemetry.MetricCollector.RAW_QUEUE_SIZE; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.api.metrics.SpanMetricRegistryImpl; +import datadog.trace.api.telemetry.CoreMetricCollector.CoreMetric; +import java.util.Collection; +import org.junit.jupiter.api.Test; + +/** + * Regression guard for the queue-full delta-loss bug in {@link + * CoreMetricCollector#prepareMetrics()}. + * + *

Before the fix, the drain read a counter with {@code getValueAndReset()} — which clears the + * counter's delta baseline — and only then tried to {@code offer()} it onto a fixed-capacity queue. + * When the queue was already full the offer failed, so that already-reset delta was lost + * permanently rather than being carried to the next cycle. The fix checks {@code + * remainingCapacity()} before reading each counter and stops early, leaving the untouched counters + * for next time. + */ +class CoreMetricCollectorQueueFullTest { + + /** + * Fills the drain past its queue capacity and verifies that every counter's delta is reported + * exactly once across as many cycles as it takes to empty — none dropped while the queue is full. + * + *

{@link CoreMetricCollector} is a singleton over the global {@link SpanMetricRegistryImpl}, + * so this test can't assume a clean registry. It stays robust by tagging its own instrumentations + * with a unique prefix and counting only those, and it drains cycle-by-cycle until the queue + * yields nothing at all (so it never terminates early just because a cycle happened to surface + * only unrelated counters left behind by other tests). With the fix the sum of our own metrics + * equals the number we created; with the bug the ones skipped while the queue was full are lost, + * so the sum comes up short. + */ + @Test + void skippedCountersRetainDeltaAcrossCycles() { + SpanMetricRegistryImpl registry = SpanMetricRegistryImpl.getInstance(); + CoreMetricCollector collector = CoreMetricCollector.getInstance(); + + // Discard anything left in the queue by other tests sharing the singleton. + collector.drain(); + + // Unique instrumentation-name prefix so we count only the instrumentations this test creates, + // regardless of whatever else has already polluted the shared registry. CoreMetricCollector + // emits these as "integration_name:" tags. + String namePrefix = "queue-full-test-" + System.nanoTime() + "-"; + String tagPrefix = "integration_name:" + namePrefix; + + // Create more non-zero counters than the queue can hold so at least one collection cycle + // overflows. A single onSpanCreated() per instrumentation leaves exactly one non-zero counter + // (spans_created) each; the rest read as 0 and are skipped. + int created = RAW_QUEUE_SIZE + 200; + for (int i = 0; i < created; i++) { + registry.get(namePrefix + i).onSpanCreated(); + } + + // Drain cycle-by-cycle until the queue is fully emptied, summing how many of our own metrics + // surface. Guard the loop so a bug that never converges fails loudly instead of hanging. + int seen = 0; + int cycles = 0; + while (true) { + collector.prepareMetrics(); + Collection drained = collector.drain(); + if (drained.isEmpty()) { + break; + } + for (CoreMetric metric : drained) { + if (isOurs(metric, tagPrefix)) { + seen++; + } + } + assertTrue(++cycles < 20, "drain did not converge; possible lost/looping delta"); + } + + assertEquals( + created, + seen, + "every counter's delta must be reported exactly once; a full queue must not drop any"); + } + + private static boolean isOurs(CoreMetric metric, String tagPrefix) { + for (String tag : metric.tags) { + if (tag.startsWith(tagPrefix)) { + return true; + } + } + return false; + } +}