Guard core-metric telemetry drain against queue-full delta loss

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

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()}.
+ *
+ * <p>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.
+   *
+   * <p>{@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:<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<CoreMetric> 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;
+  }
+}