fix(pipeline): stage one chunk per trace in a single request#159
Conversation
Overall package sizeSelf size: 30.06 MB Dependency sizes| name | version | self size | total size | |------|---------|-----------|------------|🤖 This report was automatically generated by heaviest-objects-in-the-universe |
prepareChunk overwrote the single prepared chunk on each call, so a flush carrying spans from multiple traces could only send the last one. Callers worked around that by passing every span to one prepareChunk, which lumps distinct trace_ids into a single chunk: flush_chunk then treats them as one segment and stamps trace-level tags (sampling priority, _dd.p.dm, origin, top_level) onto only the first span, corrupting sampling and grouping under load. Stage chunks in a Vec instead: prepareChunk pushes one chunk per call and sendPreparedChunk sends them all as a single multi-trace request. The exporter calls prepareChunk once per trace, so each chunk is one segment with correct per-trace tags. send_trace_chunks_async already accepts multiple chunks, so the request shape is unchanged.
bengl
force-pushed
the
bengl/pipeline-accumulate-chunks
branch
from
f50c693 to
76ee33a
Compare
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 76ee33a58d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if len == 0 { | ||
| // Nothing to send: drop any previously prepared-but-unsent chunk so | ||
| // a caller that ignores this `false` cannot later resend a stale one. | ||
| if let Some(old_spans) = self.prepared_spans.borrow_mut().take() { | ||
| self.cbs.borrow_mut().recycle_spans(old_spans); | ||
| } | ||
| // Nothing to prepare for this trace; leave any chunks already staged | ||
| // for other traces in this same flush untouched. | ||
| return Ok(false); |
There was a problem hiding this comment.
Because the staged buffer is no longer cleared on the no-span path, a previously prepared chunk survives a later prepareChunk(0, ...) call. In the back-pressure/skip path that this code previously handled, or when a caller treats the returned false as a no-op reset, the next sendPreparedChunk() will drain and send those abandoned spans even though the last prepare reported that there was nothing to send, so stale traces can be emitted with a later flush.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Thanks — but this is a false positive for the accumulate model, and clearing on empty would actually be a regression here.
Two things:
-
No caller passes
len == 0. The onlyprepareChunkcaller is the exporter'sflushSpansGrouped, which skips empty groups before it ever calls in (if (!spanIds || spanIds.length === 0) continue), and groups are built by bucketing pending spans per trace so each has ≥1 span. Thelen == 0arm is purely defensive; nothing exercises it. -
Clearing on empty would drop valid pending traces. The old single-chunk model treated
prepareChunk(0)as "reset the one staged chunk." In the new model the staging buffer holds one chunk per trace for the current flush, so a later empty prepare must leave the siblings alone — wiping them would discard other traces that are legitimately staged for this same request. That's why the arm is intentionally a no-op rather than a reset.
The only way a staged chunk goes unsent is if a later prepareChunk in the same loop throws; the next flush's sendPreparedChunk drains everything via mem::take, so those are real extracted spans sent one request late — bounded, not stale/duplicated.
Aaalibaba42
left a comment
Aaalibaba42
left a comment
There was a problem hiding this comment.
That makes sense, to me at least
2 participants
What
Stage one chunk per trace in the WASM pipeline so a flush carrying spans from multiple traces sends them as a correct multi-trace request.
prepareChunkused to store a singleOption<Vec<Span>>and overwrite it on every call, andsendPreparedChunksent exactly that one chunk. To send more than one trace, the JS exporter passed all spans (many trace_ids) in oneprepareChunk— soflush_chunktreated them as a single segment and stamped trace-level tags (sampling priority,_dd.p.dm, origin,top_level) onto only the first span.Why
Under load, the native-spans exporter batches spans from many traces into one flush (spans pile up while an async send is in flight). Decoding the agent payload showed the impact: hundreds of distinct traces lumped into one chunk per request, and only ~1 span per request carrying a sampling priority. That corrupts trace grouping and sampling downstream (the agent can't sample/group traces it can't tell apart), effectively losing traces under load even though the raw spans are delivered.
Change
prepared_spansbecomesVec<Vec<Span>>— a stack of per-trace chunks.prepareChunkpushes its chunk (one per call); empty results are not staged.sendPreparedChunkdrains all staged chunks (std::mem::take) and sends them together —send_trace_chunks_asyncalready acceptsVec<Vec<Span>>, so the outgoing request shape is unchanged (still one HTTP request per flush).The paired dd-trace-js change groups the flush batch by trace and calls
prepareChunkonce per trace (each a single segment with correct per-trace tags), then onesendPreparedChunk.Test
New
test/pipeline.jscase: two spans in two distinct traces, prepared as two chunks, must arrive as two separate trace chunks (outer msgpack array length 2). Fails under the old overwrite behavior. Full pipeline suite: 46/46.Notes