Skip to content

Run client-side stats through WorkQueue #12339

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
dougqh wants to merge 7 commits into
base: dougqh/apmlp-1642-queue-api
Choose a base branch
from
Draft

Run client-side stats through WorkQueue #12339

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
99c449b
Trial: run client-side stats through WorkQueue
dougqh Aug 27, 2026
8fd5357
Trial: bind one producer per aggregator instead of one per trace
dougqh Aug 27, 2026
877682d
Trial: carry the hoisted peer tag schema as a second context
dougqh Aug 27, 2026
0ef82d3
Trial: drain the inbox in one batched pass again
dougqh Aug 27, 2026
1db932b
Trial: rename the drain barrier to isDrained
dougqh Aug 27, 2026
abf85e8
Trial: let the inbox hold the stopped state instead of the drainer
dougqh Aug 28, 2026
a812d47
Trial: split the publish loop into a decision pass and an admission pass
dougqh Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 26 additions & 16 deletions dd-trace-core/src/main/java/datadog/trace/common/metrics/Aggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -18,7 +19,7 @@ final class Aggregator implements Runnable {

private static final Logger log = LoggerFactory.getLogger(Aggregator.class);

private final MessagePassingQueue<InboxItem> inbox;
private final WorkQueue<InboxItem> inbox;
private final AggregateTable aggregates;
private final MetricWriter writer;
private final HealthMetrics healthMetrics;
Expand All @@ -45,7 +46,7 @@ final class Aggregator implements Runnable {

Aggregator(
MetricWriter writer,
MessagePassingQueue<InboxItem> inbox,
WorkQueue<InboxItem> inbox,
int maxAggregates,
long reportingInterval,
TimeUnit reportingIntervalTimeUnit,
Expand All @@ -66,7 +67,7 @@ final class Aggregator implements Runnable {

Aggregator(
MetricWriter writer,
MessagePassingQueue<InboxItem> inbox,
WorkQueue<InboxItem> inbox,
int maxAggregates,
long reportingInterval,
TimeUnit reportingIntervalTimeUnit,
Expand Down Expand Up @@ -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) {
Expand All @@ -111,9 +114,11 @@ public void run() {
log.debug("metrics aggregator exited");
}

private final class Drainer implements MessagePassingQueue.Consumer<InboxItem> {

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<InboxItem> {

@Override
public void accept(InboxItem item) {
Expand All @@ -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
Expand All @@ -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) {
Expand Down
98 changes: 72 additions & 26 deletions dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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;

Expand Down Expand Up @@ -77,7 +78,7 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList

private final Set<String> ignoredResources;
private final Thread thread;
private final MessagePassingQueue<InboxItem> inbox;
private final WorkQueue<InboxItem> inbox;
private final Sink sink;
private final MetricWriter metricWriter;
private final Aggregator aggregator;
Expand Down Expand Up @@ -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.
*
* <p>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<CoreSpan<?>, 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.
*
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand All @@ -372,8 +391,11 @@ public Future<Boolean> 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);
Expand All @@ -394,7 +416,12 @@ public Future<Boolean> forceReport() {
public boolean publish(List<? extends CoreSpan<?>> 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
Expand All @@ -406,21 +433,38 @@ public boolean publish(List<? extends CoreSpan<?>> 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<? extends CoreSpan<?>> 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;
}
Expand All @@ -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;
Expand Down Expand Up @@ -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. */
Expand Down Expand Up @@ -679,7 +720,7 @@ public void stop() {
if (null != cancellation) {
cancellation.cancel();
}
inbox.offer(STOP);
inbox.tryPut(STOP);
}

@Override
Expand Down Expand Up @@ -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);
}
}

Expand All @@ -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;
}
}
Loading