Skip to content

Run client-side stats through WorkQueue - #12339

Draft
dougqh wants to merge 7 commits into
dougqh/apmlp-1642-queue-apifrom
dougqh/apmlp-1642-client-stats-trial
Draft

Run client-side stats through WorkQueue#12339
dougqh wants to merge 7 commits into
dougqh/apmlp-1642-queue-apifrom
dougqh/apmlp-1642-client-stats-trial

Conversation

@dougqh

@dougqh dougqh commented Aug 29, 2026

Copy link
Copy Markdown
Contributor

Stacked on #12313. Trial branch — not a merge candidate as it stands. Its purpose is to give #12313 a real caller, so the API is reviewed against use rather than on its own terms.

Client-side stats is currently the only prospective adopter of WorkQueue anywhere in the tree: every other queue user (TraceProcessingWorker, SpanSamplingWorker, PendingTraceBuffer, OkHttpSink, DefaultDataStreamsMonitoring, LLMObsIntakeWorker, BuildIdCollector, both feature-flagging writers) imports only Queues or MessagePassingBlockingQueue, i.e. the raw jctools factory. So this branch is the entirety of the evidence that #12313's surface is usable.

What it exercises

WorkQueue member exercised here
tryPut(element) ✔ control signals (STOP / REPORT / CLEAR)
tryPutBatch(source, context, producer) ✔ the publish path
process(limit, consumer) ✔ the aggregator drain loop
close / isClosed / size / dropped
tryPutBatch(Collection), tryPutBatch(T...)
RejectHandler overload
tryReserve / Reservation
RetryStrategy / RetryQueue / MaxRetries, processOrRetry / processOrHandle

The unexercised rows are the honest reason this PR exists.

The commits

  1. Run client-side stats through WorkQueue — replaces the hand-rolled jctools inbox.
  2. Bind one producer per aggregator instead of one per tracesnapshotProducer becomes a non-capturing bound-once field, so admission allocates nothing.
  3. Carry the hoisted peer tag schema as a second context — the schema is read once per trace, not once per span; this is what BiContextualProducer exists for.
  4. Drain the inbox in one batched pass again — restores the old jctools drain(size()) shape.
  5. Rename the drain barrier to isDrained.
  6. Let the inbox hold the stopped state instead of the drainerDrainer.stopped is deleted; STOP calls inbox.close(), and the run loop and drainer guards read that one flag. Also makes publish(List) ask isClosed() once per trace before statsExportEnabled(), so post-shutdown publishing costs one volatile read instead of a capacity's worth of snapshots built for a consumer that has already exited.
  7. Split the publish loop into a decision pass and an admission pass — the one to look at. See below.

On the last commit: it works, and I don't think it's an improvement

publish(List) was one fused loop doing three jobs — filter spans, accumulate counted / forceKeep, admit. tryPutBatch means the queue owns the admission walk, and the queue stops walking when it runs out of room. Two of those jobs need every eligible span regardless of admission, and the ignored-resource case is a break that a producer returning null cannot express. So using tryPutBatch at all forces two passes; there is no lighter version.

What it gets right: ineligible spans are declined (the producer returns null), which is neither an admission nor a drop — so counted - admitted is exactly the eligible spans that did not fit, which is precisely what onStatsInboxFull() wants. A rejected-elements return could not have given that, because a place is claimed before the producer is asked, so a full queue cannot tell a genuine refusal from an element the producer would have declined. That is why #12313's transforming form returns a count.

What it costs: +34/−22 for the same three jobs. Twelve fused lines become thirty unfused ones, plus limit/position index bookkeeping, a subList ternary, a loop to report a metric that used to be one inline call, and a duplicated eligibility test in snapshotIfEligible that has to stay in agreement with the first pass with nothing enforcing it. The fusion was the compression.

Latent, not a regression: inbox.dropped() now over-counts, because an ineligible span past the fill point claims a place and fails before the producer can decline it. Nothing reads dropped() in dd-trace-core today.

So commit 7 is offered as a thing to look at, not a thing to agree with. Reverting it leaves commits 1–6, which stand on their own.

Verification

:dd-trace-core:test --tests 'datadog.trace.common.metrics.*' — 151 tests, 0 failures, no test changed by commit 7. :utils:queue-utils:test — 84 tests. spotlessCheck clean on both modules.

@dougqh dougqh added tag: ai generated Largely based on code generated by an AI or LLM tag: no release notes Changes to exclude from release notes comp: metrics Metrics type: refactoring labels Aug 29, 2026
@dougqh
dougqh force-pushed the dougqh/apmlp-1642-client-stats-trial branch from 21142fa to e8ccf73 Compare August 29, 2026 01:36
@datadog-official

This comment has been minimized.

@dougqh
dougqh force-pushed the dougqh/apmlp-1642-client-stats-trial branch from e8ccf73 to c87ef0b Compare August 29, 2026 02:04
dougqh and others added 7 commits August 28, 2026 22:19
Converts the ClientStatsAggregator inbox from a raw jctools MPSC queue to
WorkQueue<InboxItem>, 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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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.
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.
@dougqh
dougqh force-pushed the dougqh/apmlp-1642-client-stats-trial branch from c87ef0b to a812d47 Compare August 29, 2026 02:21
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

comp: metrics Metrics tag: ai generated Largely based on code generated by an AI or LLM tag: no release notes Changes to exclude from release notes type: refactoring

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant