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..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 @@ -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, @@ -95,11 +96,13 @@ AggregateTable aggregates() { public void run() { Thread currentThread = Thread.currentThread(); Drainer drainer = new Drainer(); - while (!currentThread.isInterrupted() && !drainer.stopped) { + while (!currentThread.isInterrupted() && !inbox.isClosed()) { try { - if (!inbox.isEmpty()) { - inbox.drain(drainer); - } else { + // 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) { @@ -111,9 +114,11 @@ public void run() { log.debug("metrics aggregator exited"); } - private final class Drainer implements MessagePassingQueue.Consumer { - - boolean stopped = false; + /** + * 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 { @Override public void accept(InboxItem item) { @@ -132,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 @@ -143,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 858aa041b2c..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 @@ -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.BiContextualProducer; +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; @@ -110,6 +111,24 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList */ private volatile PeerTagSchema cachedPeerTagSchema; + /** + * 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::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. * @@ -274,7 +293,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 +365,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) { @@ -372,8 +391,11 @@ public Future forceReport() { // Try to send the report signal ReportSignal reportSignal = new ReportSignal(); boolean published = false; - while (thread.isAlive() && !published) { - published = inbox.offer(reportSignal); + // 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 { Thread.sleep(10); @@ -394,7 +416,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 @@ -406,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, isTopLevel, 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; } @@ -432,13 +476,14 @@ private boolean shouldComputeMetric(CoreSpan span, boolean isTopLevel) { && span.getDurationNano() > 0; } - private boolean publish(CoreSpan span, boolean isTopLevel, PeerTagSchema peerTagSchema) { + /** + * 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 SpanSnapshot snapshot(CoreSpan span, PeerTagSchema peerTagSchema) { + boolean isTopLevel = span.isTopLevel(); 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()) { - healthMetrics.onStatsInboxFull(); - return error; - } // Extract HTTP method and endpoint only if the feature is enabled String httpMethod = null; String httpEndpoint = null; @@ -503,11 +548,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 +720,7 @@ public void stop() { if (null != cancellation) { cancellation.cancel(); } - inbox.offer(STOP); + inbox.tryPut(STOP); } @Override @@ -732,7 +773,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); } } @@ -744,8 +785,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() { - return inbox.isEmpty(); + boolean isDrained() { + 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..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,8 +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; @@ -15,10 +19,13 @@ 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 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 { @@ -31,10 +38,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 +55,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 +65,96 @@ 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(); + } + + /** + * 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); 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); } }