Skip to content

Commit 56590b8

Browse files
mccullsclaude
mcculls
and
claude
committed
Limit growth of OTLP trace buffer
* Proactively flush OTLP traces between scheduled flushes to try to keep payload size below 5 MiB * Add hard limit of 64 MiB to OTLP payloads, as recommended in https://opentelemetry.io/docs/specs/otlp/ Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
1 parent af2d214 commit 56590b8

11 files changed

Lines changed: 229 additions & 36 deletions

File tree

‎components/json/src/main/java/datadog/json/JsonWriter.java‎

Lines changed: 40 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ public final class JsonWriter implements Flushable, AutoCloseable {
1919
private final JsonStructure structure;
2020

2121
private boolean requireComma;
22+
private int bytesWritten;
2223

2324
/** Creates a writer with structure check. */
2425
public JsonWriter() {
@@ -245,6 +246,11 @@ public void flush() {
245246
}
246247
}
247248

249+
/** Returns the approximate number of bytes written so far; exact for ASCII content. */
250+
public int sizeInBytes() {
251+
return this.bytesWritten;
252+
}
253+
248254
@Override
249255
public void close() {
250256
try {
@@ -267,78 +273,84 @@ private void endsValue() {
267273

268274
private void write(char ch) {
269275
try {
270-
this.writer.write(ch);
276+
doWrite(ch);
271277
} catch (IOException ignored) {
272278
}
273279
}
274280

275281
private void writeStringLiteral(String str) {
276282
try {
277-
this.writer.write('"');
283+
doWrite('"');
278284

279285
for (int i = 0; i < str.length(); ++i) {
280286
char c = str.charAt(i);
281287
// Escape any char outside ASCII to their Unicode equivalent
282288
if (c > 127) {
283-
this.writer.write('\\');
284-
this.writer.write('u');
285-
this.writer.write(HEX_DIGITS[(c >>> 12) & 0xF]);
286-
this.writer.write(HEX_DIGITS[(c >>> 8) & 0xF]);
287-
this.writer.write(HEX_DIGITS[(c >>> 4) & 0xF]);
288-
this.writer.write(HEX_DIGITS[c & 0xF]);
289+
doWrite('\\');
290+
doWrite('u');
291+
doWrite(HEX_DIGITS[(c >>> 12) & 0xF]);
292+
doWrite(HEX_DIGITS[(c >>> 8) & 0xF]);
293+
doWrite(HEX_DIGITS[(c >>> 4) & 0xF]);
294+
doWrite(HEX_DIGITS[c & 0xF]);
289295
} else {
290296
switch (c) {
291297
case '"': // Quotation mark
292298
case '\\': // Reverse solidus
293299
case '/': // Solidus
294-
this.writer.write('\\');
295-
this.writer.write(c);
300+
doWrite('\\');
301+
doWrite(c);
296302
break;
297303
case '\b': // Backspace
298-
this.writer.write('\\');
299-
this.writer.write('b');
304+
doWrite('\\');
305+
doWrite('b');
300306
break;
301307
case '\f': // Form feed
302-
this.writer.write('\\');
303-
this.writer.write('f');
308+
doWrite('\\');
309+
doWrite('f');
304310
break;
305311
case '\n': // Line feed
306-
this.writer.write('\\');
307-
this.writer.write('n');
312+
doWrite('\\');
313+
doWrite('n');
308314
break;
309315
case '\r': // Carriage return
310-
this.writer.write('\\');
311-
this.writer.write('r');
316+
doWrite('\\');
317+
doWrite('r');
312318
break;
313319
case '\t': // Horizontal tab
314-
this.writer.write('\\');
315-
this.writer.write('t');
320+
doWrite('\\');
321+
doWrite('t');
316322
break;
317323
default:
318324
if (c < 0x20) {
319-
this.writer.write('\\');
320-
this.writer.write('u');
321-
this.writer.write('0');
322-
this.writer.write('0');
323-
this.writer.write(HEX_DIGITS[(c >>> 4) & 0xF]);
324-
this.writer.write(HEX_DIGITS[c & 0xF]);
325+
doWrite('\\');
326+
doWrite('u');
327+
doWrite('0');
328+
doWrite('0');
329+
doWrite(HEX_DIGITS[(c >>> 4) & 0xF]);
330+
doWrite(HEX_DIGITS[c & 0xF]);
325331
} else {
326-
this.writer.write(c);
332+
doWrite(c);
327333
}
328334
break;
329335
}
330336
}
331337
}
332338

333-
this.writer.write('"');
339+
doWrite('"');
334340
} catch (IOException ignored) {
335341
}
336342
}
337343

338344
private void writeStringRaw(String str) {
339345
try {
340346
this.writer.write(str);
347+
this.bytesWritten += str.length(); // exact if ASCII, estimate otherwise
341348
} catch (IOException ignored) {
342349
}
343350
}
351+
352+
private void doWrite(char ch) throws IOException {
353+
this.writer.write(ch);
354+
this.bytesWritten++; // exact if ASCII, estimate otherwise
355+
}
344356
}

‎components/json/src/test/java/datadog/json/JsonWriterTest.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -165,6 +165,18 @@ void testCompleteObject() {
165165
}
166166
}
167167

168+
@Test
169+
void testSizeInBytes() {
170+
try (JsonWriter writer = new JsonWriter()) {
171+
assertEquals(0, writer.sizeInBytes(), "Check empty writer size");
172+
173+
writer.beginObject().name("string").value("café").endObject();
174+
175+
assertEquals(
176+
writer.toByteArray().length, writer.sizeInBytes(), "Check size matches written bytes");
177+
}
178+
}
179+
168180
@Test
169181
void testCompleteArray() {
170182
try (JsonWriter writer = new JsonWriter()) {

‎dd-trace-core/src/main/java/datadog/trace/common/writer/OtlpPayloadDispatcher.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99
import java.util.List;
1010

1111
final class OtlpPayloadDispatcher implements PayloadDispatcher {
12+
private static final int FLUSH_THRESHOLD_BYTES = 5 << 20; // 5 MiB
13+
1214
private final OtlpTraceCollector collector;
1315
private final OtlpSender sender;
1416

@@ -20,6 +22,10 @@ final class OtlpPayloadDispatcher implements PayloadDispatcher {
2022
@Override
2123
public void addTrace(List<? extends CoreSpan<?>> trace) {
2224
collector.addTrace(trace);
25+
// flush proactively to keep payload size bounded
26+
if (collector.sizeInBytes() >= FLUSH_THRESHOLD_BYTES) {
27+
flush();
28+
}
2329
}
2430

2531
@Override

‎dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpProtoBuffer.java‎

Lines changed: 30 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -18,12 +18,23 @@
1818
* @see GrowableBuffer
1919
*/
2020
public final class OtlpProtoBuffer {
21+
// hard limit to avoid unbounded buffering; matches OTLP spec's recommended default
22+
public static final int MAX_CAPACITY_BYTES = 64 << 20; // 64 MiB
23+
2124
private final int initialCapacity;
2225
private ByteBuffer buffer;
2326
private int remaining;
2427

2528
public OtlpProtoBuffer(int requiredCapacity) {
2629
this.initialCapacity = nextPowerOfTwo(requiredCapacity);
30+
if (this.initialCapacity > MAX_CAPACITY_BYTES) {
31+
throw new IllegalArgumentException(
32+
"OTLP payload initial capacity of "
33+
+ this.initialCapacity
34+
+ " bytes exceeds maximum buffer size of "
35+
+ MAX_CAPACITY_BYTES
36+
+ " bytes");
37+
}
2738
this.buffer = ByteBuffer.allocate(initialCapacity);
2839
this.remaining = initialCapacity;
2940
}
@@ -96,6 +107,11 @@ public ByteBuffer flip() {
96107
return buffer;
97108
}
98109

110+
/** Returns the number of bytes currently recorded in the buffer. */
111+
public int sizeInBytes() {
112+
return buffer.capacity() - remaining;
113+
}
114+
99115
/**
100116
* Returns an {@link OtlpPayload} containing the protobuf encoded content.
101117
*
@@ -123,10 +139,21 @@ private void checkCapacity(int required) {
123139
ByteBuffer oldBuffer = flip();
124140
int oldSize = oldBuffer.remaining();
125141
// round up to next multiple of initialCapacity that can accommodate required
126-
int newSize = (oldSize + required + initialCapacity - 1) & -initialCapacity;
127-
ByteBuffer newBuffer = ByteBuffer.allocate(newSize);
142+
// (uses long arithmetic so overflow can be detected before allocating)
143+
long newSize = ((long) oldSize + required + initialCapacity - 1) & -initialCapacity;
144+
if (newSize > MAX_CAPACITY_BYTES) {
145+
throw new IllegalStateException(
146+
"OTLP payload exceeds maximum buffer size of "
147+
+ MAX_CAPACITY_BYTES
148+
+ " bytes: "
149+
+ oldSize
150+
+ " bytes buffered, "
151+
+ required
152+
+ " more requested");
153+
}
154+
ByteBuffer newBuffer = ByteBuffer.allocate((int) newSize);
128155
// copy over old content so it stays at the far end
129-
remaining = newSize - oldSize;
156+
remaining = (int) newSize - oldSize;
130157
newBuffer.position(remaining);
131158
newBuffer.put(oldBuffer);
132159
buffer = newBuffer;

‎dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceCollector.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,9 @@ public abstract class OtlpTraceCollector {
1515
/** Collects all spans added since the last collection. */
1616
public abstract OtlpPayload collectTraces();
1717

18+
/** Returns the number of bytes buffered since the last collection. */
19+
public abstract int sizeInBytes();
20+
1821
protected final boolean shouldExport(CoreSpan<?> span) {
1922
return span.samplingPriority() > 0 // trace-level sampling priority
2023
|| span.getTag(SPAN_SAMPLING_MECHANISM_TAG) != null; // span-level sampling priority

‎dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceJsonCollector.java‎

Lines changed: 19 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
import static datadog.trace.core.otlp.common.OtlpCommonJson.writeScopeAndSchema;
44
import static datadog.trace.core.otlp.common.OtlpPayload.JSON_CONTENT_TYPE;
5+
import static datadog.trace.core.otlp.common.OtlpProtoBuffer.MAX_CAPACITY_BYTES;
56
import static datadog.trace.core.otlp.common.OtlpResourceJson.TRACE_RESOURCE_FRAGMENT;
67
import static datadog.trace.core.otlp.trace.OtlpTraceJson.writeSpan;
78

@@ -54,11 +55,22 @@ public void addTrace(List<? extends CoreSpan<?>> spans) {
5455
payloadStarted = true;
5556
}
5657

57-
for (CoreSpan<?> span : spans) {
58-
visitSpan(span);
58+
try {
59+
for (CoreSpan<?> span : spans) {
60+
visitSpan(span);
61+
}
62+
} catch (Throwable e) {
63+
// reset the buffer for subsequent traces
64+
stop();
65+
throw e;
5966
}
6067
}
6168

69+
@Override
70+
public int sizeInBytes() {
71+
return writer == null ? 0 : writer.sizeInBytes();
72+
}
73+
6274
/**
6375
* Marshals the traces collected so far into a JSON payload.
6476
*
@@ -178,5 +190,10 @@ private void completeSpan() {
178190
// reset temporary elements for next span
179191
currentSpan = null;
180192
currentSpanLinks = Collections.emptyList();
193+
194+
if (writer.sizeInBytes() > MAX_CAPACITY_BYTES) {
195+
throw new IllegalStateException(
196+
"OTLP payload exceeds maximum buffer size of " + MAX_CAPACITY_BYTES + " bytes");
197+
}
181198
}
182199
}

‎dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java‎

Lines changed: 14 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -56,12 +56,23 @@ public void addTrace(List<? extends CoreSpan<?>> spans) {
5656
payloadStarted = true;
5757
}
5858

59-
// OtlpProtoBuffer collects spans in reverse
60-
for (int i = spans.size() - 1; i >= 0; i--) {
61-
visitSpan(spans.get(i));
59+
try {
60+
// OtlpProtoBuffer collects spans in reverse
61+
for (int i = spans.size() - 1; i >= 0; i--) {
62+
visitSpan(spans.get(i));
63+
}
64+
} catch (Throwable e) {
65+
// reset the buffer for subsequent traces
66+
stop();
67+
throw e;
6268
}
6369
}
6470

71+
@Override
72+
public int sizeInBytes() {
73+
return protobuf.sizeInBytes();
74+
}
75+
6576
/**
6677
* Marshals the traces collected so far into a chunked payload.
6778
*

‎dd-trace-core/src/test/java/datadog/trace/common/writer/OtlpPayloadDispatcherTest.java‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
import static java.util.Collections.singletonList;
66
import static org.junit.jupiter.api.Assertions.assertEquals;
77
import static org.junit.jupiter.api.Assertions.assertTrue;
8+
import static org.mockito.Mockito.any;
89
import static org.mockito.Mockito.mock;
910
import static org.mockito.Mockito.verify;
1011
import static org.mockito.Mockito.verifyNoInteractions;
@@ -95,6 +96,39 @@ void emptyTraceForwardsNothing() {
9596
verifyNoInteractions(sender);
9697
}
9798

99+
@Test
100+
void largeTraceTriggersProactiveFlushWithoutExplicitFlush() {
101+
OtlpPayloadDispatcher dispatcher = new OtlpPayloadDispatcher(sender, collector);
102+
collector.fakeSizeInBytes = Integer.MAX_VALUE;
103+
104+
dispatcher.addTrace(singletonList(sampledSpan()));
105+
106+
// no explicit dispatcher.flush() call
107+
ArgumentCaptor<OtlpPayload> captor = ArgumentCaptor.forClass(OtlpPayload.class);
108+
verify(sender).send(captor.capture());
109+
assertEquals(1 /*spans*/, captor.getValue().getContentLength());
110+
}
111+
112+
@Test
113+
void belowFlushThresholdDoesNotTriggerProactiveFlush() {
114+
OtlpPayloadDispatcher dispatcher = new OtlpPayloadDispatcher(sender, collector);
115+
collector.fakeSizeInBytes = (5 << 20) - 1; // one byte under FLUSH_THRESHOLD_BYTES
116+
117+
dispatcher.addTrace(singletonList(sampledSpan()));
118+
119+
verifyNoInteractions(sender);
120+
}
121+
122+
@Test
123+
void atFlushThresholdTriggersProactiveFlush() {
124+
OtlpPayloadDispatcher dispatcher = new OtlpPayloadDispatcher(sender, collector);
125+
collector.fakeSizeInBytes = 5 << 20; // exactly FLUSH_THRESHOLD_BYTES
126+
127+
dispatcher.addTrace(singletonList(sampledSpan()));
128+
129+
verify(sender).send(any(OtlpPayload.class));
130+
}
131+
98132
@Test
99133
void getApisIsEmpty() {
100134
OtlpPayloadDispatcher dispatcher = new OtlpPayloadDispatcher(sender, collector);
@@ -143,6 +177,9 @@ private static CoreSpan<?> unsetSpan() {
143177
private static class TestCollector extends OtlpTraceCollector {
144178
final List<CoreSpan<?>> spansToExport = new ArrayList<>();
145179

180+
// lets tests drive the proactive-flush threshold independently of spansToExport
181+
int fakeSizeInBytes;
182+
146183
@Override
147184
public void addTrace(List<? extends CoreSpan<?>> spans) {
148185
for (CoreSpan<?> span : spans) {
@@ -165,5 +202,10 @@ public OtlpPayload collectTraces() {
165202
spansToExport.clear();
166203
}
167204
}
205+
206+
@Override
207+
public int sizeInBytes() {
208+
return fakeSizeInBytes;
209+
}
168210
}
169211
}

‎dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpProtoBufferTest.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -60,6 +60,14 @@ void initialPayloadIsEmpty() {
6060
assertEquals("application/x-protobuf", payload.getContentType());
6161
}
6262

63+
@Test
64+
void constructorRejectsCapacityExceedingMaxCapacity() {
65+
// rounds up to a power of two above MAX_CAPACITY_BYTES; must reject before allocating
66+
assertThrows(
67+
IllegalArgumentException.class,
68+
() -> new OtlpProtoBuffer(OtlpProtoBuffer.MAX_CAPACITY_BYTES + 1));
69+
}
70+
6371
// ─── recordMessage(GrowableBuffer, int) ──────────────────────────────────
6472

6573
@Test

0 commit comments

Comments
 (0)