[SVLS-9716] Split oversized trace payloads into smaller chunks - #154
Conversation
kathiehuang
changed the title
There was a problem hiding this comment.
Pull request overview
This PR addresses oversized trace payload handling in datadog-trace-agent by splitting protobuf v0.7 tracer payloads along trace-chunk boundaries and ensuring the aggregator can still flush a single oversized payload without blocking the queue.
Changes:
- Split v0.7
pb::TracerPayloadinstances recursively by chunk boundary to stay withinMAX_CONTENT_SIZE_BYTESwhen possible, and warn when a single chunk still exceeds the limit. - Update
TraceAggregator::get_batch()to flush a single oversized payload standalone instead of re-queuing indefinitely. - Add unit/integration tests covering payload splitting and oversized flush behavior; add
prostdependency for encoded-size measurement.
Reviewed changes
Copilot reviewed 3 out of 4 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| crates/datadog-trace-agent/src/trace_processor.rs | Add protobuf-encoded size based splitting for oversized v0.7 payloads; warn on un-splittable oversized chunks; add tests validating splitting and send behavior. |
| crates/datadog-trace-agent/src/aggregator.rs | Ensure an oversized single payload can be flushed alone so it doesn’t block batching; add regression test. |
| crates/datadog-trace-agent/Cargo.toml | Add prost dependency needed for encoded size calculations. |
| Cargo.lock | Lockfile updates for prost. |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
a472237 to
37c57fc
Compare
|
@codex review |
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![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: 37c57fcfeb
ℹ️ 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".
| fn encoded_size(payloads: &[pb::TracerPayload]) -> usize { | ||
| payloads.iter().map(Message::encoded_len).sum() |
There was a problem hiding this comment.
When a V07 tracer payload is within a few bytes of MAX_CONTENT_SIZE_BYTES, summing only TracerPayload::encoded_len() undercounts the actual request body because serialization adds length-delimited protobuf framing around each payload. The splitter can therefore leave a nominally fitting piece unchanged, and the aggregator also uses this underestimated size when coalescing pieces, producing an HTTP body over the intake limit that may be rejected. Compute the size of the complete serialized V07 payload, including its envelope/framing, or reserve that overhead before comparing against the limit.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Added a safety buffer of 64 bytes in 6da92f8 as each outgoing request wraps the Vec<TracerPayload> in a pb::AgentPayload envelope.
Computing the size of the complete serialized V07 payload would require making private functions public or creating a new public helper function in libdd-trace-utils. SendData::len() and coalesce_send_data also both mention that size is an approximation.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 4 changed files in this pull request and generated 1 comment.
Suppressed comments (1)
crates/datadog-trace-agent/src/trace_processor.rs:285
- For V07,
sizeis computed as the innerTracerPayloadprotobuf size, but the split threshold subtractsV07_ENVELOPE_OVERHEAD_BYTES. Passing the pre-overhead size intoSendData::new(and warning only whensize > MAX_CONTENT_SIZE_BYTES) can undercount queued payloads, allowing a coalesced batch to exceedMAX_CONTENT_SIZE_BYTESonce the outerAgentPayloadenvelope is added (and also misses the intended warn forsplit_budget < size <= MAX_CONTENT_SIZE_BYTES).
let pieces: Vec<(TracerPayloadCollection, usize)> = match payload {
TracerPayloadCollection::V07(payloads) => {
let split_budget =
MAX_CONTENT_SIZE_BYTES.saturating_sub(V07_ENVELOPE_OVERHEAD_BYTES);
split_oversized_payloads(payloads, split_budget)
.into_iter()
.map(|tp| {
let size = encoded_size(std::slice::from_ref(&tp));
(TracerPayloadCollection::V07(vec![tp]), size)
})
.collect()
}
other => vec![(other, body_size)],
};
if pieces.len() > 1 {
debug!(
piece_count = pieces.len(),
"Oversized trace payload split into multiple pieces"
);
}
for (piece, size) in pieces {
let send_data = SendData::new(
size,
piece,
tracer_header_tags.clone(),
&config.trace_intake,
);
if size > MAX_CONTENT_SIZE_BYTES {
warn!(
payload_size = size,
max_content_size_bytes = MAX_CONTENT_SIZE_BYTES,
"Trace chunk is over max batch size; sending standalone"
);
}
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e7e0559225
ℹ️ 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".
| split_oversized_payloads(payloads, split_budget) | ||
| .into_iter() | ||
| .map(|tp| { | ||
| let size = encoded_size(std::slice::from_ref(&tp)); |
There was a problem hiding this comment.
Although the 64-byte split budget was added, the current code still records only the inner TracerPayload length in each SendData. TraceAggregator::get_batch sums those lengths, and trace_flusher.rs then passes the batch to coalesce_send_data, which wraps multiple payloads in one AgentPayload; consequently, a near-limit split piece followed by another fitting payload can have inner lengths at or below the limit while the serialized request exceeds it due to per-payload framing. Include the framing allowance in the queued size (for every piece), or measure the complete coalesced encoding before applying the limit.
Useful? React with 👍 / 👎.
|
@codex review |
|
Codex Review: Didn't find any major issues. More of your lovely PRs please. Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
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". |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 3 out of 4 changed files in this pull request and generated no new comments.
Suppressed comments (1)
Previously missed (1) — in code that hasn't changed since the last review.
crates/datadog-trace-agent/src/trace_processor.rs:262
- In the V07 path,
split_oversized_payloads(...)is applied unconditionally and each resultingTracerPayloadis wrapped into its ownTracerPayloadCollection::V07(vec![tp]). This changes the non-oversized case too: ifcollect_pb_trace_chunksreturns multipleTracerPayloads, they will now be queued/sent as separateSendDataitems even when the combined encoded size fits underMAX_CONTENT_SIZE_BYTES, increasing queue entries and likely increasing downstream flush request count.
Consider preserving the previous batching behavior when the total V07 encoded size is within the budget, and only splitting/fan-out when the total would exceed MAX_CONTENT_SIZE_BYTES.
let pieces: Vec<(TracerPayloadCollection, usize)> = match payload {
TracerPayloadCollection::V07(payloads) => {
let split_budget =
MAX_CONTENT_SIZE_BYTES.saturating_sub(V07_ENVELOPE_OVERHEAD_BYTES);
split_oversized_payloads(payloads, split_budget)
| /// Splits `payloads` so that each returned `TracerPayload`'s encoded size fits within | ||
| /// `max_size` where possible. Recursively bisects by trace-chunk boundary. A single chunk | ||
| /// that's still oversized is returned as-is and gets sent standalone. | ||
| fn split_oversized_payloads( |
There was a problem hiding this comment.
It looks like this splitting logic differs slightly from that of the go agent. This will split down to as many groups of chunks as needed to get each group under the ~3.2MB limit whereas the go agent will split into a maximum of 4 groups of chunks. The implementation is more precise, every piece is guaranteed to fit unless it's a single indivisible chunk, but it also means the number of groups here is unbounded rather than capped at 4. How costly is calling encoded_len() at every level of the recursion?
There was a problem hiding this comment.
Each encoded_len() call does a traversal of the payload's spans/tags/meta to compute the size, proportional to how much data is in it. Since we call it once per recursive invocation, and each level of the recursion still walks its (shrinking) share of the data:
- Level 0: 1 call, cost O(N) (N = total payload size)
- Level 1: 2 calls, each on ~N/2 → total O(N)
- Level 2: 4 calls, each on ~N/4 → total O(N)
- ...continuing for O(log C) levels (C = chunk count)
This makes the total cost O(N log C).
The recursion only splits a branch further if that branch's own encoded_len() is still over budget, however—and since this is the first time we've seen this issue it looks like the overwhelming majority of traffic is well under 3.2MB and needs zero splitting at all. But if we wanted to improve this, we could compute each chunk's encoded_len() once up front and then use prefix sums to get any contiguous range's total size via subtraction, which would make this O(N)
There was a problem hiding this comment.
I think it's probably fine given that the payloads being split are all between 3.2MB and 10MB, correct? So in the worse case there should only be a few levels of recursion. I just wanted to call out the difference between this implementation and the 4 chunk maximum that the go agent uses.
| if let Err(err) = tx.send(send_data).await { | ||
| return log_and_create_http_response( | ||
| &format!("Error sending traces to the trace flusher: {err}"), | ||
| StatusCode::INTERNAL_SERVER_ERROR, | ||
| ); | ||
| } |
There was a problem hiding this comment.
If tx.send() fails on group 3 of 5, groups 1–2 are already enqueued for delivery and will still get flushed normally. But the client only sees a single 500 for the whole request. Will that cause a retry for the entire original, unsplit, payload and cause groups 1–2 to be sent twice?
There was a problem hiding this comment.
Great point! It's true that the client will only see a single 500 for the whole request if that happens. Whether a retry happens depends on the tracer. I checked each one:
| Tracer | Retries on 5xx? | Mechanism |
|---|---|---|
| dd-trace-dotnet | Yes | Api.cs — SendWithRetry re-sends the same in-memory buffer up to 5x with exponential backoff for any 5xx (excludes 429/413/408, which are deliberately never retried). |
| dd-trace-js | No | request.js's onResponse treats any non-2xx as terminal (complete(error, ...)) — retry (handleError) is wired only to socket-level 'error' events, gated to a fixed set of network error codes (ECONNRESET, ETIMEDOUT, etc.), never HTTP status. |
| dd-trace-py | No | _send_payload always returns a Response object even for a 5xx; the fibonacci_backoff_with_jitter wrapper only retries when the call raises (connection-level exceptions), not on status code. Logged via log.error, counted as http.dropped.*. |
| dd-trace-java | No | DDAgentApi.sendSerializedTraces converts any non-200 straight into a terminal Response.failed(...). No retry logic anywhere in the writer package (confirmed via grep). Payload is dropped, logged, metrics incremented. |
So the duplicate-delivery risk from a 500 on a split payload only exists for dd-trace-dotnet.
I looked at what the Go agent does and its HTTP receiver replies 200 OK to the tracer as soon as it successfully decodes the payload, before splitting. If we wanted to match that we would reply to the tracer right after collect_pb_trace_chunks (after the payload is decoded, validated, and enriched), and the splitting + tx.send() loop would move into a tokio::spawn so that it's decoupled from the response.
There was a problem hiding this comment.
I looked at what the Go agent does and its HTTP receiver replies 200 OK to the tracer as soon as it successfully decodes the payload, before splitting. If we wanted to match that we would reply to the tracer right after collect_pb_trace_chunks (after the payload is decoded, validated, and enriched), and the splitting + tx.send() loop would move into a tokio::spawn so that it's decoupled from the response.
That sounds like the correct approach to me! We should avoid a scenario where duplicate traces could be sent.
There was a problem hiding this comment.
Implemented in 1f1c919 - I ended up splitting & building Vec<SendData> before the spawn and moving only the tx.send() loop over that already-built Vec<SendData> into the tokio::spawn.
However, this move has spawned a couple comments #154 (comment)
There was a problem hiding this comment.
🟡 Changes recommended
process_traces now returns HTTP success before confirming the enqueue sends succeeded (and without backpressure), which can acknowledge trace ingestion even when the flusher channel is closed or blocked.
Once you've addressed the issues Copilot identified, you can request another Copilot review.
Review details
- Files reviewed: 3/4 changed files
- Comments generated: 1
- Review effort level: Lite
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1f1c91998c
ℹ️ 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".
| tokio::spawn(async move { | ||
| for send_data in send_datas { | ||
| if let Err(err) = tx.send(send_data).await { | ||
| error!("Error sending traces to the trace flusher: {err}"); | ||
| return; | ||
| } | ||
| } | ||
| }); |
There was a problem hiding this comment.
When the ten-slot trace channel fills—for example, while trace_flusher.rs::flush holds the aggregator mutex during a slow or retrying intake request—each subsequent HTTP request now spawns another detached task retaining its entire send_datas vector and immediately returns 200. This bypasses the bounded channel's backpressure, allowing clients to keep submitting payloads and causing unbounded memory growth during an intake slowdown; previously the awaited tx.send kept the handler pending. Await the enqueue operation or route all pieces through a bounded, lifecycle-managed producer before reporting success.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
route all pieces through a bounded, lifecycle-managed producer before reporting success
This would involve gating the number of detached enqueue tasks with a bounded Semaphore, sized to the channel capacity (10 permits, so worst-case backlog ≈ 2× the channel's own bound instead of unbounded):
pub struct ServerlessTraceProcessor {
pub stats_concentrator: Option<StatsConcentratorHandle>,
enqueue_permits: Arc<tokio::sync::Semaphore>, // added semaphore field, initialized with 10 permits in mini_agent.rs
}
...
let permit = self.enqueue_permits.clone().acquire_owned().await; // acquire an owned permit
...
tokio::spawn(async move { // move the permit into the spawned task
let _permit = permit; // releases the permit when this task ends
for send_data in send_datas {
if let Err(err) = tx.send(send_data).await {
error!("Error sending traces to the trace flusher: {err}");
return;
}
}
});
acquire_owned()'s .await resolves instantly if a permit is free, but if all 10 are already checked out by still-in-flight enqueue tasks, this .await - which happens before we spawn/reply - suspends the request handler itself until one frees up. So under saturation, a new requests's reply is delayed exactly like it would've been under the old design where tx.send() blocked the handler, except now gated on "are 10 enqueue attempts already outstanding" instead of "is the channel itself full".
@duncanpharvey Do you think this is necessary or do you think we can trust that the traces can get consistently and continuously flushed?
There was a problem hiding this comment.
I think the bounded semaphore is needed since it protects against unbounded memory growth in the trace agent (exactly the issue that prompted these changes). I'd also add a timeout so a stuck flush can't hold a request open indefinitely.
The Go agent returns 429 if a feature flag is enabled, and a comment on the PR that introduced it suggests 429 was meant to become the default in a later API version, but that never happened, and Go still returns 200 by default today. I'd rather not introduce a new client visible status code (the 429 response code) as part of this PR in case that changes tracer behavior in unexpected ways.
- https://github.com/DataDog/datadog-agent/blob/db369ae2d731da603c8f84015b060af2af955931/pkg/trace/api/api.go#L181-L185
- pkg/trace/api: add a feature flag for 429 presampling responses datadog-agent#3469
In my opinion the bounded semaphore with a timeout and a 200 response would be sufficient. Some experimentation may be necessary to have confidence in a timeout value that is comfortably longer than a normal request cycle but not so long that permits stay tied up for a large fraction of a flush interval before we start shedding load. 2-3 seconds may be a good starting point.
There was a problem hiding this comment.
I see, that makes sense! Thanks for checking on the response code detail. I added the semaphore in 4336cd8 and 29fec9a similar to the Go agent's semaphore
- I did 10 permits but the Go agent derives permit count from number of CPUs
The permit acquisition happens at the start of process_traces (like the Go agent where it happens before r.decodeTracerPayload) and gets carried all the way to the spawned task at the end so that the whole decode/enrich/split/enqueue pipeline is bounded to MAX_IN_FLIGHT_ENQUEUES concurrent requests. Acquiring it only before enqueue still allows unlimited concurrent requests to decode, enrich, and retain large trace payloads while waiting
There was a problem hiding this comment.
🔵 Needs a closer look
It changes core trace ingestion/batching and request-to-enqueue concurrency behavior in ways that can have subtle operational impact and warrants careful human validation.
Review details
- Files reviewed: 7/8 changed files
- Comments generated: 0 new
- Review effort level: Lite
|
@codex review |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 29fec9a120
ℹ️ 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".
| Ok(Err(_)) | Err(_) => { | ||
| warn!("Could not acquire an enqueue permit in time; dropping traces"); | ||
| return log_and_create_traces_success_http_response( | ||
| "Dropped traces due to enqueue capacity", | ||
| StatusCode::OK, | ||
| ); |
There was a problem hiding this comment.
When all enqueue permits remain occupied for two seconds, this branch returns while body is still unread. The agent serves persistent HTTP/1 connections (mini_agent.rs:429), and the analogous stats-dropping path explicitly collects the body to keep the connection reusable (stats_processor.rs:48-53); failing to do so here can close the tracer's connection and force a reconnect whenever backpressure triggers. Collect and discard the request body before returning the load-shedding response.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Updated to collect body in trace-dropping path 52d9607 similarly to when stats get dropped
52d9607 to
11a6fad
Compare
What does this PR do?
This fixes a bug where a single oversized trace payload could permanently jam the mini-agent's flush queue. This PR proactively splits oversized payloads into smaller pieces along trace-chunk boundaries before they reach the queue. If a trace chunk by itself is oversized, send it standalone.
TraceAggregator::get_batch()can get stuck if a single queued trace payload size exceedsMAX_CONTENT_SIZE_BYTES(3.2MB) - the oversized chunk gets requeued to the front of the queue every call, blocking all subsequent traces behind it with no error logged by either the agent or tracer.This PR fixes the issue by splitting an oversized
TracerPayloadinto multiple smaller ones by recursively bisecting by trace-chunk boundary before it's queued, mirroring how the Go agent's tracev1.go splits by chunk.trace_processor.rs): binary-bisect a TracerPayload's chunks until each piece's encoded size fits under MAX_CONTENT_SIZE_BYTES, or until it can't be split further (a single chunk still over the limit is returned as-is).encoded_size()instead of msgpackbody_size(which is raw and pre-enrichment). This ensures that a payload that only becomes oversized after tag/obfuscation enrichment is not missed.V07_ENVELOPE_OVERHEAD_BYTESfor the protobufAgentPayloadenvelope overhead added when a payload is actually serialized for sending -encoded_len()on the bareTracerPayloadalone doesn't capture this.debug!when a payload splits into more than one piece andwarn!when an individual piece is still oversized after splitting and gets sent standalone.process_tracesacquires a permit from a bounded semaphore (MAX_IN_FLIGHT_ENQUEUES = 10) before decoding the request body, and carries that same permit through enrichment, splitting, and the spawnedtx.send()task at the end. If a permit doesn't free up withinenqueue_permit_timeout_secs(default 2s), the request sheds load: it drains the still-unread body, logs a warning, and replies 200 OK so tracers don't retry the request.Motivation
https://datadoghq.atlassian.net/browse/SLES-9716
Additional Notes
prostdependency for the encoded-size computationDescribe how to test/QA your changes
Added unit tests for split_oversized_payloads/split_tracer_payload/encoded_size (no-split-needed, multi-chunk-split, single-oversized-chunk-returned-as-is cases) and an integration test (test_process_trace_sends_oversized_single_chunk_standalone).
Manual e2e testing:
serverless-compat-self-monitoring branch used for testing/reproducing
I reproduced the error by creating a P0V3 Azure Function with two functions: one that sends a trace smaller than 3.2MB
undersizedhandlerand one that sends an oversized traceoversizedHandler. I deployed them with a custom-built binary of serverless compat with an extra error log at line 55:I created a cron job that curls
undersizedHandlerevery minute, and then after a few minutes, curledoversizedHandleronce. I saw this error in the logs:The
oversizedHandlertrace never made it to Datadog.I let undersizedHandler continue getting curled, and after a few hours checked metrics. Memory growth grows unbounded while traces stop.
I deployed the same Azure Function but instrumented with a custom binary built from this PR, curled
undersizedHandlerconsistently, and then curledoversizedHandleragain, and saw this warning log:I also see the
oversizedHandlertrace (link):azure.functions.invokeand 600oversized.childspansspan_ids are uniqueSubsequent traces land in Datadog as well, and memory does not grow.
I made another function that also creates an oversized trace payload but with each span getting no parent, meaning each one starts a new trace which leads to multiple trace chunks. After curling it, I saw that one of those trace payloads got split:
Link to trace
oversized_many_chunks.tracespan and 30oversized_many_chunks.childspans, all correctly parented to that root and all unique