From 99c449b324260aac41bac185ace1d8a59ca7347f Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 26 Aug 2026 22:25:31 -0400 Subject: [PATCH 1/7] Trial: run client-side stats through WorkQueue Converts the ClientStatsAggregator inbox from a raw jctools MPSC queue to WorkQueue, to see what the API costs and buys on a real caller. publish() no longer builds a SpanSnapshot it may have to throw away: the tag lookups, the peer/additional tag arrays and the snapshot itself move into a Producer the queue invokes only after reserving a slot. The racy size() >= capacity() pre-check goes with it. The producer is a mutable SnapshotRequest reused across the spans of one trace, so deferral costs one allocation per trace rather than one per span. The Aggregator drain loop becomes process(drainer, LOG_AND_DISCARD); the strategy restores the logging that the old catch(Throwable) in run() did, since a WorkQueue with no strategy discards a failed item silently. Co-Authored-By: Claude Opus 5 --- .../trace/common/metrics/Aggregator.java | 17 ++--- .../common/metrics/ClientStatsAggregator.java | 67 +++++++++++++------ .../ClientStatsAggregatorInboxFullTest.java | 55 ++++++++++++--- 3 files changed, 102 insertions(+), 37 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 4c54eaf8b42..7469fb85365 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,13 +2,14 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; +import datadog.common.queue.WorkQueue; 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; import edu.umd.cs.findbugs.annotations.SuppressFBWarnings; import java.util.concurrent.TimeUnit; -import org.jctools.queues.MessagePassingQueue; +import java.util.function.Consumer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -18,7 +19,7 @@ final class Aggregator implements Runnable { private static final Logger log = LoggerFactory.getLogger(Aggregator.class); - private final MessagePassingQueue inbox; + private final WorkQueue inbox; private final AggregateTable aggregates; private final MetricWriter writer; private final HealthMetrics healthMetrics; @@ -45,7 +46,7 @@ final class Aggregator implements Runnable { Aggregator( MetricWriter writer, - MessagePassingQueue inbox, + WorkQueue inbox, int maxAggregates, long reportingInterval, TimeUnit reportingIntervalTimeUnit, @@ -66,7 +67,7 @@ final class Aggregator implements Runnable { Aggregator( MetricWriter writer, - MessagePassingQueue inbox, + WorkQueue inbox, int maxAggregates, long reportingInterval, TimeUnit reportingIntervalTimeUnit, @@ -97,9 +98,9 @@ public void run() { Drainer drainer = new Drainer(); while (!currentThread.isInterrupted() && !drainer.stopped) { try { - if (!inbox.isEmpty()) { - inbox.drain(drainer); - } else { + // process reports whether there was an item to work on; a failing item throws out of it, + // into the same catch that has always kept one bad item from ending the drain loop. + if (!inbox.process(drainer)) { Thread.sleep(sleepMillis); } } catch (InterruptedException e) { @@ -111,7 +112,7 @@ public void run() { log.debug("metrics aggregator exited"); } - private final class Drainer implements MessagePassingQueue.Consumer { + private final class Drainer implements Consumer { boolean stopped = false; diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 858aa041b2c..835830b32eb 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -16,7 +16,9 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; -import datadog.common.queue.Queues; +import datadog.common.queue.Producer; +import datadog.common.queue.WorkQueue; +import datadog.common.queue.WorkQueues; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; @@ -39,7 +41,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; -import org.jctools.queues.MessagePassingQueue; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -77,7 +78,7 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList private final Set ignoredResources; private final Thread thread; - private final MessagePassingQueue inbox; + private final WorkQueue inbox; private final Sink sink; private final MetricWriter metricWriter; private final Aggregator aggregator; @@ -274,7 +275,7 @@ private static AdditionalTagsSchema additionalTagsSchemaFrom(Config config) { this.additionalTagsSchema = additionalTagsSchema; this.includeEndpointInMetrics = includeEndpointInMetrics; this.otlpStatsExportEnabled = metricWriter instanceof OtlpStatsMetricWriter; - this.inbox = Queues.mpscArrayQueue(queueSize); + this.inbox = WorkQueues.createMpscQueue(queueSize); this.features = features; this.healthMetrics = healthMetric; this.sink = sink; @@ -346,7 +347,7 @@ public boolean report() { boolean published; int attempts = 0; do { - published = inbox.offer(REPORT); + published = inbox.tryPut(REPORT); ++attempts; } while (!published && attempts < 10); if (!published) { @@ -373,7 +374,7 @@ public Future forceReport() { ReportSignal reportSignal = new ReportSignal(); boolean published = false; while (thread.isAlive() && !published) { - published = inbox.offer(reportSignal); + published = inbox.tryPut(reportSignal); if (!published) { try { Thread.sleep(10); @@ -406,6 +407,9 @@ public boolean publish(List> trace) { // ignoredResources is fixed for the lifetime of the aggregator and typically empty; hoist the // check so the common case skips both the lookup and the getResourceName() call per span. final boolean hasIgnoredResources = !ignoredResources.isEmpty(); + // One request per trace rather than one per span: it carries the arguments the deferred + // snapshot needs, and is dead by the time this method returns. + SnapshotRequest request = null; for (CoreSpan span : trace) { boolean isTopLevel = span.isTopLevel(); if (shouldComputeMetric(span, isTopLevel)) { @@ -417,7 +421,10 @@ public boolean publish(List> trace) { } } counted++; - forceKeep |= publish(span, isTopLevel, peerTagSchema); + if (request == null) { + request = new SnapshotRequest(); + } + forceKeep |= publish(request, span, isTopLevel, peerTagSchema); } } healthMetrics.onClientStatTraceComputed(counted, trace.size(), !forceKeep); @@ -432,13 +439,37 @@ private boolean shouldComputeMetric(CoreSpan span, boolean isTopLevel) { && span.getDurationNano() > 0; } - private boolean publish(CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { - boolean error = span.getError() > 0; - // size() is approximate on jctools MPSC queues but good enough for a fast-path overflow check. - if (inbox.size() >= inbox.capacity()) { + private boolean publish( + SnapshotRequest request, CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { + request.span = span; + request.isTopLevel = isTopLevel; + request.peerTagSchema = peerTagSchema; + // The inbox reserves a slot before calling back, so a full inbox costs nothing beyond this + // call: none of the tag lookups, no peer/additional tag arrays, no SpanSnapshot. The old racy + // size() >= capacity() pre-check is gone with it. + if (!inbox.tryPut(request)) { healthMetrics.onStatsInboxFull(); - return error; } + return span.getError() > 0; + } + + /** + * Builds a {@link SpanSnapshot} once the inbox has reserved a slot for it. Mutable and reused + * across the spans of one trace so deferral costs one allocation per trace, not one per span. + */ + private final class SnapshotRequest implements Producer { + CoreSpan span; + boolean isTopLevel; + PeerTagSchema peerTagSchema; + + @Override + public InboxItem produce() { + return snapshot(span, isTopLevel, peerTagSchema); + } + } + + private SpanSnapshot snapshot(CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { + boolean error = span.getError() > 0; // Extract HTTP method and endpoint only if the feature is enabled String httpMethod = null; String httpEndpoint = null; @@ -503,11 +534,7 @@ private boolean publish(CoreSpan span, boolean isTopLevel, PeerTagSchema peer grpcStatusCode, additionalTagValues, tagAndDuration); - if (!inbox.offer(snapshot)) { - healthMetrics.onStatsInboxFull(); - } - // force keep keys if there are errors - return error; + return snapshot; } /** Returns the first non-null span tag among {@code keys}, in order, or {@code null} if none. */ @@ -679,7 +706,7 @@ public void stop() { if (null != cancellation) { cancellation.cancel(); } - inbox.offer(STOP); + inbox.tryPut(STOP); } @Override @@ -732,7 +759,7 @@ private void disable() { // the aggregator drains existing snapshots and ships them on the next report cycle; the // sink rejects that payload and fires DOWNGRADED again, which retries disable() against a // now-empty inbox. Worst case: one extra reporting cycle of stale data. - inbox.offer(CLEAR); + inbox.tryPut(CLEAR); } } @@ -746,6 +773,6 @@ public void run(ClientStatsAggregator target) { @VisibleForTesting boolean isEmpty() { - return inbox.isEmpty(); + return inbox.size() == 0; } } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java index 3683bef7f42..c7e8d2001a8 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java @@ -4,6 +4,7 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -15,10 +16,9 @@ import org.junit.jupiter.api.Test; /** - * Coverage for the inbox-full fast-path in {@code ClientStatsAggregator.publish}: when the - * producer-side inbox is at capacity, the next {@code publish} call short-circuits before any tag - * extraction or {@code SpanSnapshot} allocation and reports {@code onStatsInboxFull()} to health - * metrics. + * Coverage for publishing into a full inbox: the {@link datadog.common.queue.WorkQueue} reserves a + * slot before it calls back, so a rejected span costs no tag extraction and no {@code SpanSnapshot} + * allocation, and the rejection is reported to health metrics. */ class ClientStatsAggregatorInboxFullTest { @@ -31,10 +31,9 @@ void publishFiresOnStatsInboxFullOnceInboxIsAtCapacity() { when(features.supportsMetrics()).thenReturn(true); when(features.peerTags()).thenReturn(Collections.emptySet()); - // Small inbox; jctools MPSC array queue rounds up to the next power of two, so use a power of + // Small inbox; the MPSC array backing rounds up to the next power of two, so use a power of // two directly. Note: we deliberately do NOT call aggregator.start() so the consumer thread - // never drains -- snapshots accumulate in the inbox until capacity, then the next publish hits - // the size-vs-capacity fast path. + // never drains -- snapshots accumulate in the inbox until capacity, then admission rejects. int queueSize = 8; ClientStatsAggregator aggregator = new ClientStatsAggregator( @@ -49,8 +48,8 @@ void publishFiresOnStatsInboxFullOnceInboxIsAtCapacity() { SECONDS, /* includeEndpointInMetrics */ false); - // Publish well past capacity. The first `queueSize` calls land in the inbox; subsequent calls - // see size >= capacity and hit the fast path. + // Publish well past capacity. The first `queueSize` calls land in the inbox; the rest are + // rejected. for (int i = 0; i < queueSize * 4; i++) { aggregator.publish(Collections.>singletonList(metricsEligibleSpan())); } @@ -59,6 +58,44 @@ void publishFiresOnStatsInboxFullOnceInboxIsAtCapacity() { aggregator.close(); } + /** The point of the reserve-first admission: a rejected span is never turned into a snapshot. */ + @Test + void publishBuildsNoSnapshotOnceInboxIsAtCapacity() { + HealthMetrics healthMetrics = mock(HealthMetrics.class); + MetricWriter writer = mock(MetricWriter.class); + Sink sink = mock(Sink.class); + DDAgentFeaturesDiscovery features = mock(DDAgentFeaturesDiscovery.class); + when(features.supportsMetrics()).thenReturn(true); + when(features.peerTags()).thenReturn(Collections.emptySet()); + + int queueSize = 8; + ClientStatsAggregator aggregator = + new ClientStatsAggregator( + Collections.emptySet(), + features, + healthMetrics, + sink, + writer, + /* maxAggregates */ 16, + queueSize, + /* reportingInterval */ 10, + SECONDS, + /* includeEndpointInMetrics */ false); + + for (int i = 0; i < queueSize; i++) { + aggregator.publish(Collections.>singletonList(metricsEligibleSpan())); + } + + // A fresh span published into the now-full inbox: only the eligibility checks should touch it. + CoreSpan rejected = metricsEligibleSpan(); + aggregator.publish(Collections.>singletonList(rejected)); + + verify(rejected, never()).getServiceName(); + verify(rejected, never()).getOperationName(); + verify(rejected, never()).getSpanKindString(); + aggregator.close(); + } + @SuppressWarnings({"rawtypes", "unchecked"}) private static CoreSpan metricsEligibleSpan() { CoreSpan span = mock(CoreSpan.class); From 8fd5357f73f075edae093793f77d7d7364eed9ba Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 26 Aug 2026 22:39:18 -0400 Subject: [PATCH 2/7] Trial: bind one producer per aggregator instead of one per trace isTopLevel is a field read on the span's context and the peer tag schema is non-null forever after bootstrap, so neither is context the producer has to be handed: the span alone is enough. The producer becomes a field bound once at construction and admission allocates nothing. Costs one extra volatile read per span, and a mid-trace schema change is now seen by the spans after it rather than by the next trace. Co-Authored-By: Claude Opus 5 --- .../common/metrics/ClientStatsAggregator.java | 43 ++++++++----------- 1 file changed, 18 insertions(+), 25 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 835830b32eb..b43538d2b56 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -16,7 +16,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; -import datadog.common.queue.Producer; +import datadog.common.queue.ContextualProducer; import datadog.common.queue.WorkQueue; import datadog.common.queue.WorkQueues; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; @@ -111,6 +111,13 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList */ private volatile PeerTagSchema cachedPeerTagSchema; + /** + * Builds the snapshot for a span once the inbox has reserved a slot for it. Bound to this + * aggregator once, at construction, so admission costs no allocation at all: the span is the + * context, and everything else the snapshot needs is reachable from one or the other. + */ + private final ContextualProducer, InboxItem> snapshotProducer = this::snapshotOf; + /** * Previous peer-tag schema, kept until the next reporting cycle. * @@ -407,9 +414,6 @@ public boolean publish(List> trace) { // ignoredResources is fixed for the lifetime of the aggregator and typically empty; hoist the // check so the common case skips both the lookup and the getResourceName() call per span. final boolean hasIgnoredResources = !ignoredResources.isEmpty(); - // One request per trace rather than one per span: it carries the arguments the deferred - // snapshot needs, and is dead by the time this method returns. - SnapshotRequest request = null; for (CoreSpan span : trace) { boolean isTopLevel = span.isTopLevel(); if (shouldComputeMetric(span, isTopLevel)) { @@ -421,10 +425,7 @@ public boolean publish(List> trace) { } } counted++; - if (request == null) { - request = new SnapshotRequest(); - } - forceKeep |= publish(request, span, isTopLevel, peerTagSchema); + forceKeep |= publish(span); } } healthMetrics.onClientStatTraceComputed(counted, trace.size(), !forceKeep); @@ -439,33 +440,25 @@ private boolean shouldComputeMetric(CoreSpan span, boolean isTopLevel) { && span.getDurationNano() > 0; } - private boolean publish( - SnapshotRequest request, CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { - request.span = span; - request.isTopLevel = isTopLevel; - request.peerTagSchema = peerTagSchema; + private boolean publish(CoreSpan span) { // The inbox reserves a slot before calling back, so a full inbox costs nothing beyond this // call: none of the tag lookups, no peer/additional tag arrays, no SpanSnapshot. The old racy // size() >= capacity() pre-check is gone with it. - if (!inbox.tryPut(request)) { + if (!inbox.tryPut(span, snapshotProducer)) { healthMetrics.onStatsInboxFull(); } return span.getError() > 0; } /** - * Builds a {@link SpanSnapshot} once the inbox has reserved a slot for it. Mutable and reused - * across the spans of one trace so deferral costs one allocation per trace, not one per span. + * Re-derives what the publish loop had already computed, rather than carrying it into the + * producer: {@code isTopLevel} is a field read on the span's context, and the peer tag schema is + * non-null forever once {@link #bootstrapPeerTagSchema()} has run, which {@link #publish(List)} + * guarantees before it reaches any span. The cost is one extra volatile read per span, and a + * schema change mid-trace is now visible to the spans after it rather than to the next trace. */ - private final class SnapshotRequest implements Producer { - CoreSpan span; - boolean isTopLevel; - PeerTagSchema peerTagSchema; - - @Override - public InboxItem produce() { - return snapshot(span, isTopLevel, peerTagSchema); - } + private InboxItem snapshotOf(CoreSpan span) { + return snapshot(span, span.isTopLevel(), cachedPeerTagSchema); } private SpanSnapshot snapshot(CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { From 877682d35c2d6531fd9f6a62d247274fbee0ecde Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 26 Aug 2026 23:58:53 -0400 Subject: [PATCH 3/7] Trial: carry the hoisted peer tag schema as a second context Reserve-first admission turns the producer inside out, and the schema the publish loop had hoisted out of the span loop did not survive the inversion -- it went back to one volatile read per span, and a schema change mid-trace became visible to the rest of that trace rather than to the next one. BiContextualProducer carries it across instead, so the read is once per trace again and the trace boundary is where it was. isTopLevel stays derived inside the producer: it is a field read on the span, so there is nothing to carry. Co-Authored-By: Claude Opus 5 --- .../common/metrics/ClientStatsAggregator.java | 31 +++++++++---------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index b43538d2b56..7d5de474fdb 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -16,7 +16,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; -import datadog.common.queue.ContextualProducer; +import datadog.common.queue.BiContextualProducer; import datadog.common.queue.WorkQueue; import datadog.common.queue.WorkQueues; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; @@ -113,10 +113,12 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList /** * Builds the snapshot for a span once the inbox has reserved a slot for it. Bound to this - * aggregator once, at construction, so admission costs no allocation at all: the span is the - * context, and everything else the snapshot needs is reachable from one or the other. + * aggregator once, at construction, so admission costs no allocation at all: the span and the + * schema the publish loop hoisted are the two contexts, and everything else the snapshot needs is + * reachable from this aggregator. */ - private final ContextualProducer, InboxItem> snapshotProducer = this::snapshotOf; + private final BiContextualProducer, PeerTagSchema, InboxItem> snapshotProducer = + this::snapshot; /** * Previous peer-tag schema, kept until the next reporting cycle. @@ -425,7 +427,7 @@ public boolean publish(List> trace) { } } counted++; - forceKeep |= publish(span); + forceKeep |= publish(span, peerTagSchema); } } healthMetrics.onClientStatTraceComputed(counted, trace.size(), !forceKeep); @@ -440,28 +442,23 @@ private boolean shouldComputeMetric(CoreSpan span, boolean isTopLevel) { && span.getDurationNano() > 0; } - private boolean publish(CoreSpan span) { + private boolean publish(CoreSpan span, PeerTagSchema peerTagSchema) { // The inbox reserves a slot before calling back, so a full inbox costs nothing beyond this // call: none of the tag lookups, no peer/additional tag arrays, no SpanSnapshot. The old racy // size() >= capacity() pre-check is gone with it. - if (!inbox.tryPut(span, snapshotProducer)) { + if (!inbox.tryPut(span, peerTagSchema, snapshotProducer)) { healthMetrics.onStatsInboxFull(); } return span.getError() > 0; } /** - * Re-derives what the publish loop had already computed, rather than carrying it into the - * producer: {@code isTopLevel} is a field read on the span's context, and the peer tag schema is - * non-null forever once {@link #bootstrapPeerTagSchema()} has run, which {@link #publish(List)} - * guarantees before it reaches any span. The cost is one extra volatile read per span, and a - * schema change mid-trace is now visible to the spans after it rather than to the next trace. + * The schema rides along as the second context so the publish loop can keep reading it once per + * trace rather than once per span, which is what it did before admission became a callback. + * {@code isTopLevel} needs no such carriage: it is a field read on the span itself. */ - private InboxItem snapshotOf(CoreSpan span) { - return snapshot(span, span.isTopLevel(), cachedPeerTagSchema); - } - - private SpanSnapshot snapshot(CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { + private SpanSnapshot snapshot(CoreSpan span, PeerTagSchema peerTagSchema) { + boolean isTopLevel = span.isTopLevel(); boolean error = span.getError() > 0; // Extract HTTP method and endpoint only if the feature is enabled String httpMethod = null; From 0ef82d32241c75d48a1ca5a38f863e190def74a3 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Thu, 27 Aug 2026 00:26:30 -0400 Subject: [PATCH 4/7] Trial: drain the inbox in one batched pass again The move to WorkQueue had turned the aggregator's drain into one item per loop iteration, re-testing the interrupt flag and the stopped flag between each. process(limit, consumer) puts the loop back on the queue's side. The limit is whatever size() reports at the top of the pass, which is the old jctools drain semantics: take what is there, and let anything that arrives mid-pass be the next pass's work. size() is O(1) on this queue, so asking costs a read. Co-Authored-By: Claude Opus 5 --- .../java/datadog/trace/common/metrics/Aggregator.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 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 7469fb85365..f80049b0b1c 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 @@ -98,9 +98,11 @@ public void run() { Drainer drainer = new Drainer(); while (!currentThread.isInterrupted() && !drainer.stopped) { try { - // process reports whether there was an item to work on; a failing item throws out of it, - // into the same catch that has always kept one bad item from ending the drain loop. - if (!inbox.process(drainer)) { + // Take what is there in one pass, the way the old jctools drain did. size() is O(1) on + // this queue, so asking costs a read, and anything that arrives mid-pass is simply the + // next pass's work. A failing item throws out of process, into the same catch that has + // always kept one bad item from ending the drain loop. + if (inbox.process(Math.max(1, inbox.size()), drainer) == 0) { Thread.sleep(sleepMillis); } } catch (InterruptedException e) { From 1db932b285be595668cc93fffb5e20f241aa5642 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Thu, 27 Aug 2026 00:29:39 -0400 Subject: [PATCH 5/7] Trial: rename the drain barrier to isDrained size() counts capacity in use, which includes a place claimed but not yet filled, so "empty" is no longer a question it can answer. "Drained" is, and it is what the test barrier was actually waiting for. Co-Authored-By: Claude Opus 5 --- .../trace/common/metrics/ClientStatsAggregator.java | 7 ++++++- .../trace/common/metrics/ClientStatsAggregatorTest.java | 8 ++++---- 2 files changed, 10 insertions(+), 5 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 7d5de474fdb..766deb9bd6e 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -761,8 +761,13 @@ public void run(ClientStatsAggregator target) { } } + /** + * Whether the aggregator thread has taken everything the inbox held. Named for consumption rather + * than contents: {@code size()} counts capacity in use, which includes any place claimed but not + * yet filled, so "drained" is the question this can actually answer. + */ @VisibleForTesting - boolean isEmpty() { + boolean isDrained() { return inbox.size() == 0; } } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorTest.java index 6d4ccf64f19..e4ef1546ff8 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorTest.java @@ -81,7 +81,7 @@ void shouldIgnoreTracesWithNoMeasuredSpans() throws Exception { Collections.singletonList( new SimpleSpan("", "", "", "", false, false, false, 0, 0, HTTP_OK))); - waitUntilAggregatorIsEmpty(aggregator); + waitUntilAggregatorIsDrained(aggregator); clearInvocations(sink); aggregator.forceReport().get(2, SECONDS); @@ -2010,7 +2010,7 @@ void whenNoAggregateIsUpdatedInReportingIntervalNothingIsReported() throws Excep verify(writer, times(1)).finishBucket(); // second cycle - no updates at all - waitUntilAggregatorIsEmpty(aggregator); + waitUntilAggregatorIsDrained(aggregator); clearInvocations(writer); aggregator.forceReport().get(2, SECONDS); @@ -2858,10 +2858,10 @@ void cardinalityLimitsResetBetweenReportCycles() throws Exception { } } - private void waitUntilAggregatorIsEmpty(ClientStatsAggregator aggregator) + private void waitUntilAggregatorIsDrained(ClientStatsAggregator aggregator) throws InterruptedException { int i = 0; - while (!aggregator.isEmpty() && i++ < 100) { + while (!aggregator.isDrained() && i++ < 100) { Thread.sleep(10); } } From abf85e8fb083ee89a9274298481022f0b03621be Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 28 Aug 2026 16:06:34 -0400 Subject: [PATCH 6/7] Trial: let the inbox hold the stopped state instead of the drainer The drainer kept a private boolean that meant "STOP has been taken", and the run loop read it to leave. WorkQueue already models that state: close() stops admission and keeps what is queued readable, which is exactly what STOP means here. So STOP closes the inbox, and the loop and the drainer both read the one flag rather than two that have to agree. Producers can see it, which is the point. Before, the stopped state was invisible on the publish side: after the consumer exited, producers went on building a capacity's worth of snapshots for a queue nobody was draining, and then reported inbox-full for the rest of the process. Now publish asks once per trace and does nothing else. Also stops forceReport() sleeping in 10ms steps through the window between STOP being taken and the thread finishing its exit. --- .../trace/common/metrics/Aggregator.java | 23 ++++--- .../common/metrics/ClientStatsAggregator.java | 16 ++++- .../ClientStatsAggregatorInboxFullTest.java | 65 ++++++++++++++++++- 3 files changed, 91 insertions(+), 13 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 f80049b0b1c..3456a2a0778 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 @@ -96,7 +96,7 @@ AggregateTable aggregates() { public void run() { Thread currentThread = Thread.currentThread(); Drainer drainer = new Drainer(); - while (!currentThread.isInterrupted() && !drainer.stopped) { + while (!currentThread.isInterrupted() && !inbox.isClosed()) { try { // Take what is there in one pass, the way the old jctools drain did. size() is O(1) on // this queue, so asking costs a read, and anything that arrives mid-pass is simply the @@ -114,10 +114,12 @@ public void run() { log.debug("metrics aggregator exited"); } + /** + * Stateless. Whether the aggregator has stopped is the inbox's closed flag, which the run loop + * reads as its own exit condition -- see the {@link StopSignal} branch below. + */ private final class Drainer implements Consumer { - boolean stopped = false; - @Override public void accept(InboxItem item) { if (item == ClearSignal.CLEAR) { @@ -135,7 +137,7 @@ public void accept(InboxItem item) { // re-aggregated, and flushed on the next report -- where the agent rejects them again, // triggering another DOWNGRADED -> disable() -> CLEAR cycle. Worst case: one extra // reporting cycle of wasted work, which we accept for the safety of preserving STOP. - if (!stopped) { + if (!inbox.isClosed()) { aggregates.clear(); // Clear dirty too -- without this, the next report() would see dirty=true, run // expungeStaleAggregates against the (now-empty) table, find isEmpty()=true, and skip @@ -146,16 +148,21 @@ public void accept(InboxItem item) { ((SignalItem) item).complete(); } else if (item instanceof SignalItem) { SignalItem signal = (SignalItem) item; - if (!stopped) { + if (!inbox.isClosed()) { report(wallClockTime(), signal); - stopped = item instanceof StopSignal; - if (stopped) { + if (item instanceof StopSignal) { + // Closing the inbox *is* stopping. It refuses further admission, so producers stop + // building snapshots for a consumer that is on its way out, and it is the condition + // this loop already reads to leave -- one piece of state rather than two that have to + // agree. Deliberately not shutdown(): anything queued behind STOP stays readable, and + // the batch this call sits in keeps being walked. + inbox.close(); signal.complete(); } } else { signal.ignore(); } - } else if (item instanceof SpanSnapshot && !stopped) { + } else if (item instanceof SpanSnapshot && !inbox.isClosed()) { SpanSnapshot snapshot = (SpanSnapshot) item; AggregateEntry entry = aggregates.findOrInsert(snapshot); if (entry != null) { diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 766deb9bd6e..231fbee0807 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -382,7 +382,10 @@ public Future forceReport() { // Try to send the report signal ReportSignal reportSignal = new ReportSignal(); boolean published = false; - while (thread.isAlive() && !published) { + // isClosed() as well as isAlive(): the inbox closes the moment STOP is taken, which is + // strictly before the thread finishes exiting, so this stops sleeping through that window + // waiting on a signal that can no longer be admitted. + while (thread.isAlive() && !inbox.isClosed() && !published) { published = inbox.tryPut(reportSignal); if (!published) { try { @@ -404,7 +407,12 @@ public Future forceReport() { public boolean publish(List> trace) { boolean forceKeep = false; int counted = 0; - if (statsExportEnabled()) { + // Closed before enabled: closed is a volatile read, where statsExportEnabled() can reach into + // feature discovery. Once the aggregator thread has taken STOP the inbox refuses everything, + // so asking once here is the whole publish path after shutdown -- rather than a capacity's + // worth of snapshots built for a consumer that has already exited, followed by an unbounded + // run of inbox-full reports for a queue nobody is draining. + if (!inbox.isClosed() && statsExportEnabled()) { // Producer-side fast path: one volatile read and use whatever schema is currently cached. // The aggregator thread keeps this schema in sync with feature discovery in // resetCardinalityHandlers(). The only producer-side rebuild is the one-time bootstrap on @@ -446,6 +454,10 @@ private boolean publish(CoreSpan span, PeerTagSchema peerTagSchema) { // The inbox reserves a slot before calling back, so a full inbox costs nothing beyond this // call: none of the tag lookups, no peer/additional tag arrays, no SpanSnapshot. The old racy // size() >= capacity() pre-check is gone with it. + // + // A refusal is read as capacity rather than closure because the caller checked isClosed() once + // for the trace. A close landing mid-trace mis-attributes that trace's remaining spans, which + // is worth not re-reading the flag per span. if (!inbox.tryPut(span, peerTagSchema, snapshotProducer)) { healthMetrics.onStatsInboxFull(); } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java index c7e8d2001a8..84fc6bdf31a 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/ClientStatsAggregatorInboxFullTest.java @@ -2,9 +2,12 @@ import static java.util.concurrent.TimeUnit.SECONDS; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.Mockito.atLeastOnce; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -16,9 +19,13 @@ import org.junit.jupiter.api.Test; /** - * Coverage for publishing into a full inbox: the {@link datadog.common.queue.WorkQueue} reserves a - * slot before it calls back, so a rejected span costs no tag extraction and no {@code SpanSnapshot} - * allocation, and the rejection is reported to health metrics. + * Coverage for publishing into an inbox that refuses admission, which happens for two reasons: it + * is at capacity, or the aggregator has stopped and closed it. + * + *

Either way the {@link datadog.common.queue.WorkQueue} reserves a slot before it calls back, so + * a rejected span costs no tag extraction and no {@code SpanSnapshot} allocation. A capacity + * refusal is reported to health metrics; a closed inbox is not reported at all, because publishing + * to a stopped aggregator is not a pressure signal. */ class ClientStatsAggregatorInboxFullTest { @@ -96,6 +103,58 @@ void publishBuildsNoSnapshotOnceInboxIsAtCapacity() { aggregator.close(); } + /** + * Once the aggregator thread has taken STOP the inbox is closed, and that closed flag is the only + * thing publish reads for the whole trace -- no cached-schema read, no per-span eligibility + * checks, no health-metric traffic. + * + *

What this pins is the reason the stopped state lives in the queue rather than beside it. + * When it was a private flag on the drainer, producers could not see it: they went on building a + * capacity's worth of snapshots for a consumer that had already exited, and then reported + * inbox-full for the rest of the process lifetime against a queue nobody was draining. + */ + @Test + void publishTouchesNothingOnceTheAggregatorHasStopped() { + HealthMetrics healthMetrics = mock(HealthMetrics.class); + MetricWriter writer = mock(MetricWriter.class); + Sink sink = mock(Sink.class); + DDAgentFeaturesDiscovery features = mock(DDAgentFeaturesDiscovery.class); + when(features.supportsMetrics()).thenReturn(true); + when(features.peerTags()).thenReturn(Collections.emptySet()); + + ClientStatsAggregator aggregator = + new ClientStatsAggregator( + Collections.emptySet(), + features, + healthMetrics, + sink, + writer, + /* maxAggregates */ 16, + /* queueSize */ 8, + /* reportingInterval */ 10, + SECONDS, + /* includeEndpointInMetrics */ false); + + // start() so a thread is there to take STOP; close() posts it and joins, so by the time close() + // returns the thread has exited -- and the only way out of the run loop is inbox.close(). The + // inbox is empty and the drain loop sleeps 10ms, well inside the 800ms join. + aggregator.start(); + aggregator.close(); + + // Anything the lifecycle itself recorded is not the subject. + reset(healthMetrics); + + CoreSpan afterStop = metricsEligibleSpan(); + aggregator.publish(Collections.>singletonList(afterStop)); + + // getDurationNano is part of the eligibility check, which sits inside the loop: if it was not + // called, the refusal happened once for the trace rather than once per span. + verify(afterStop, never()).getDurationNano(); + verify(afterStop, never()).getServiceName(); + verify(healthMetrics, never()).onStatsInboxFull(); + verify(healthMetrics, never()).onClientStatTraceComputed(anyInt(), anyInt(), anyBoolean()); + } + @SuppressWarnings({"rawtypes", "unchecked"}) private static CoreSpan metricsEligibleSpan() { CoreSpan span = mock(CoreSpan.class); From a812d47b87e9628802a79bb63b90bf240851f289 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 28 Aug 2026 19:46:13 -0400 Subject: [PATCH 7/7] Trial: split the publish loop into a decision pass and an admission pass Uses tryPutBatch, which means the trace is walked twice, because the queue owns the admission walk and stops when it runs out of room. counted and forceKeep are about the trace and need every eligible span; the ignored resource case is a break, and a producer returning null can skip an element but cannot stop a walk. So pass one answers all three -- and finds where the break landed, which is what pass two is handed as a subList. Exact where it matters: ineligible spans are declined rather than dropped, so counted minus admitted is precisely the eligible spans that did not fit, which is what onStatsInboxFull() wants and what a rejected-elements return could not have given. Costs a second eligibility test per span, four field reads, to avoid materialising a filtered list per trace. What it does not buy is brevity: twelve fused lines become thirty unfused ones, plus index bookkeeping and a duplicated predicate in snapshotIfEligible. Kept on the trial branch to be looked at rather than assumed either way. Note inbox.dropped() now over-counts: an ineligible span past the fill point claims a place and fails before the producer can decline it. Nothing reads dropped() today, so this is latent. --- .../common/metrics/ClientStatsAggregator.java | 56 +++++++++++-------- 1 file changed, 34 insertions(+), 22 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 231fbee0807..55de8f17ec2 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -112,13 +112,22 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList private volatile PeerTagSchema cachedPeerTagSchema; /** - * Builds the snapshot for a span once the inbox has reserved a slot for it. Bound to this - * aggregator once, at construction, so admission costs no allocation at all: the span and the - * schema the publish loop hoisted are the two contexts, and everything else the snapshot needs is - * reachable from this aggregator. + * Builds the snapshot for a span once the inbox has reserved a place for it, or declines the span + * if it is not one metrics are computed for. Bound to this aggregator once, at construction, so + * admission costs no allocation at all: the span and the schema the publish loop hoisted are the + * two contexts, and everything else the snapshot needs is reachable from this aggregator. + * + *

The eligibility test runs here as well as in the first pass of {@link #publish(List)}, + * because the inbox walks the trace itself and has no way to be handed a filtered view of it + * without materialising one. Four field reads a second time is the price of not allocating a list + * per trace, and the reason the decision pass can stay separate from the admission pass. */ private final BiContextualProducer, PeerTagSchema, InboxItem> snapshotProducer = - this::snapshot; + this::snapshotIfEligible; + + private InboxItem snapshotIfEligible(CoreSpan span, PeerTagSchema peerTagSchema) { + return shouldComputeMetric(span, span.isTopLevel()) ? snapshot(span, peerTagSchema) : null; + } /** * Previous peer-tag schema, kept until the next reporting cycle. @@ -424,21 +433,38 @@ public boolean publish(List> trace) { // ignoredResources is fixed for the lifetime of the aggregator and typically empty; hoist the // check so the common case skips both the lookup and the getResourceName() call per span. final boolean hasIgnoredResources = !ignoredResources.isEmpty(); + // Two passes, because this loop answers two questions with different scopes. counted and + // forceKeep are about the trace: every eligible span contributes, whether or not the inbox + // had room for it. Admission is about the inbox: only the spans there was room for. Fused + // into one pass they read as one thing; separated, each pass says what it is for. + int limit = trace.size(); + int position = 0; for (CoreSpan span : trace) { - boolean isTopLevel = span.isTopLevel(); - if (shouldComputeMetric(span, isTopLevel)) { + if (shouldComputeMetric(span, span.isTopLevel())) { if (hasIgnoredResources) { final CharSequence resourceName = span.getResourceName(); if (resourceName != null && ignoredResources.contains(resourceName.toString())) { // skip publishing all children + limit = position; break; } } counted++; - forceKeep |= publish(span, peerTagSchema); + forceKeep |= span.getError() > 0; } + position++; } healthMetrics.onClientStatTraceComputed(counted, trace.size(), !forceKeep); + // The inbox reserves a place before calling back, so a full inbox costs nothing beyond the + // claim: none of the tag lookups, no peer/additional tag arrays, no SpanSnapshot. Ineligible + // spans are declined by the producer, which is not a drop and not counted as admitted -- + // so counted minus admitted is exactly the eligible spans that did not fit. + List> admissible = + limit == trace.size() ? trace : trace.subList(0, limit); + int admitted = inbox.tryPutBatch(admissible, peerTagSchema, snapshotProducer); + for (int refused = counted - admitted; refused > 0; refused--) { + healthMetrics.onStatsInboxFull(); + } } return forceKeep; } @@ -450,20 +476,6 @@ private boolean shouldComputeMetric(CoreSpan span, boolean isTopLevel) { && span.getDurationNano() > 0; } - private boolean publish(CoreSpan span, PeerTagSchema peerTagSchema) { - // The inbox reserves a slot before calling back, so a full inbox costs nothing beyond this - // call: none of the tag lookups, no peer/additional tag arrays, no SpanSnapshot. The old racy - // size() >= capacity() pre-check is gone with it. - // - // A refusal is read as capacity rather than closure because the caller checked isClosed() once - // for the trace. A close landing mid-trace mis-attributes that trace's remaining spans, which - // is worth not re-reading the flag per span. - if (!inbox.tryPut(span, peerTagSchema, snapshotProducer)) { - healthMetrics.onStatsInboxFull(); - } - return span.getError() > 0; - } - /** * The schema rides along as the second context so the publish loop can keep reading it once per * trace rather than once per span, which is what it did before admission became a callback.