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>

Author: mcculls

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

@@ -19,6 +19,7 @@ public final class JsonWriter implements Flushable, AutoCloseable {
   private final JsonStructure structure;
 
   private boolean requireComma;
+  private int bytesWritten;
 
   /** Creates a writer with structure check. */
   public JsonWriter() {
@@ -245,6 +246,11 @@ public void flush() {
     }
   }
 
+  /** Approximate number of bytes written so far. */
+  public int sizeInBytes() {
+    return this.bytesWritten;
+  }
+
   @Override
   public void close() {
     try {
@@ -268,13 +274,15 @@ private void endsValue() {
   private void write(char ch) {
     try {
       this.writer.write(ch);
+      this.bytesWritten++;
     } catch (IOException ignored) {
     }
   }
 
   private void writeStringLiteral(String str) {
     try {
       this.writer.write('"');
+      int count = 1;
 
       for (int i = 0; i < str.length(); ++i) {
         char c = str.charAt(i);
@@ -286,33 +294,40 @@ private void writeStringLiteral(String str) {
           this.writer.write(HEX_DIGITS[(c >>> 8) & 0xF]);
           this.writer.write(HEX_DIGITS[(c >>> 4) & 0xF]);
           this.writer.write(HEX_DIGITS[c & 0xF]);
+          count += 6;
         } else {
           switch (c) {
             case '"': // Quotation mark
             case '\\': // Reverse solidus
             case '/': // Solidus
               this.writer.write('\\');
               this.writer.write(c);
+              count += 2;
               break;
             case '\b': // Backspace
               this.writer.write('\\');
               this.writer.write('b');
+              count += 2;
               break;
             case '\f': // Form feed
               this.writer.write('\\');
               this.writer.write('f');
+              count += 2;
               break;
             case '\n': // Line feed
               this.writer.write('\\');
               this.writer.write('n');
+              count += 2;
               break;
             case '\r': // Carriage return
               this.writer.write('\\');
               this.writer.write('r');
+              count += 2;
               break;
             case '\t': // Horizontal tab
               this.writer.write('\\');
               this.writer.write('t');
+              count += 2;
               break;
             default:
               if (c < 0x20) {
@@ -322,22 +337,28 @@ private void writeStringLiteral(String str) {
                 this.writer.write('0');
                 this.writer.write(HEX_DIGITS[(c >>> 4) & 0xF]);
                 this.writer.write(HEX_DIGITS[c & 0xF]);
+                count += 6;
               } else {
                 this.writer.write(c);
+                count += 1;
               }
               break;
           }
         }
       }
 
       this.writer.write('"');
+      count += 1;
+
+      this.bytesWritten += count;
     } catch (IOException ignored) {
     }
   }
 
   private void writeStringRaw(String str) {
     try {
       this.writer.write(str);
+      this.bytesWritten += str.length(); // exact if ASCII, estimate otherwise
     } catch (IOException ignored) {
     }
   }

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

@@ -165,6 +165,54 @@ void testCompleteObject() {
     }
   }
 
+  @Test
+  void testSizeInBytes() {
+    try (JsonWriter writer = new JsonWriter()) {
+      assertEquals(0, writer.sizeInBytes(), "Check empty writer size");
+
+      writer.beginArray();
+      assertSizeInBytes(writer, "Check size after plain ASCII string");
+      writer.value("bar");
+
+      assertSizeInBytes(writer, "Check size after escaped characters");
+      writer.value("\"\\/");
+
+      assertSizeInBytes(writer, "Check size after named escapes");
+      writer.value("\b\f\n\r\t");
+
+      assertSizeInBytes(writer, "Check size after control character escape");
+      writer.value("\u0001");
+
+      assertSizeInBytes(writer, "Check size after non-ASCII character escape");
+      writer.value("café");
+
+      assertSizeInBytes(writer, "Check size after int value");
+      writer.value(3);
+
+      assertSizeInBytes(writer, "Check size after long value");
+      writer.value(3456789123L);
+
+      assertSizeInBytes(writer, "Check size after float value");
+      writer.value(3.142f);
+
+      assertSizeInBytes(writer, "Check size after double value");
+      writer.value(PI);
+
+      assertSizeInBytes(writer, "Check size after boolean value");
+      writer.value(true);
+
+      assertSizeInBytes(writer, "Check size after null value");
+      writer.nullValue();
+
+      writer.endArray();
+      assertSizeInBytes(writer, "Check final size matches written bytes");
+    }
+  }
+
+  private static void assertSizeInBytes(JsonWriter writer, String message) {
+    assertEquals(writer.toByteArray().length, writer.sizeInBytes(), message);
+  }
+
   @Test
   void testCompleteArray() {
     try (JsonWriter writer = new JsonWriter()) {

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

@@ -7,8 +7,14 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 final class OtlpPayloadDispatcher implements PayloadDispatcher {
+  private static final Logger log = LoggerFactory.getLogger(OtlpPayloadDispatcher.class);
+
+  private static final int FLUSH_THRESHOLD_BYTES = 5 << 20; // 5 MiB
+
   private final OtlpTraceCollector collector;
   private final OtlpSender sender;
 
@@ -20,13 +26,21 @@ final class OtlpPayloadDispatcher implements PayloadDispatcher {
   @Override
   public void addTrace(List<? extends CoreSpan<?>> trace) {
     collector.addTrace(trace);
+    // flush proactively to keep payload size bounded
+    if (collector.sizeInBytes() >= FLUSH_THRESHOLD_BYTES) {
+      flush();
+    }
   }
 
   @Override
   public void flush() {
-    OtlpPayload payload = collector.collectTraces();
-    if (payload != OtlpPayload.EMPTY) {
-      sender.send(payload);
+    try {
+      OtlpPayload payload = collector.collectTraces();
+      if (payload != OtlpPayload.EMPTY) {
+        sender.send(payload);
+      }
+    } catch (Throwable e) {
+      log.debug("Failed to send OTLP payload", e);
     }
   }
 

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

@@ -18,12 +18,23 @@
  * @see GrowableBuffer
  */
 public final class OtlpProtoBuffer {
+  // hard limit to avoid unbounded buffering; matches OTLP spec's recommended default
+  public static final int MAX_CAPACITY_BYTES = 64 << 20; // 64 MiB
+
   private final int initialCapacity;
   private ByteBuffer buffer;
   private int remaining;
 
   public OtlpProtoBuffer(int requiredCapacity) {
     this.initialCapacity = nextPowerOfTwo(requiredCapacity);
+    if (this.initialCapacity > MAX_CAPACITY_BYTES) {
+      throw new IllegalArgumentException(
+          "OTLP payload initial capacity of "
+              + this.initialCapacity
+              + " bytes exceeds maximum buffer size of "
+              + MAX_CAPACITY_BYTES
+              + " bytes");
+    }
     this.buffer = ByteBuffer.allocate(initialCapacity);
     this.remaining = initialCapacity;
   }
@@ -96,6 +107,11 @@ public ByteBuffer flip() {
     return buffer;
   }
 
+  /** Returns the number of bytes currently recorded in the buffer. */
+  public int sizeInBytes() {
+    return buffer.capacity() - remaining;
+  }
+
   /**
    * Returns an {@link OtlpPayload} containing the protobuf encoded content.
    *
@@ -123,10 +139,21 @@ private void checkCapacity(int required) {
       ByteBuffer oldBuffer = flip();
       int oldSize = oldBuffer.remaining();
       // round up to next multiple of initialCapacity that can accommodate required
-      int newSize = (oldSize + required + initialCapacity - 1) & -initialCapacity;
-      ByteBuffer newBuffer = ByteBuffer.allocate(newSize);
+      // (uses long arithmetic so overflow can be detected before allocating)
+      long newSize = ((long) oldSize + required + initialCapacity - 1) & -initialCapacity;
+      if (newSize > MAX_CAPACITY_BYTES) {
+        throw new IllegalStateException(
+            "OTLP payload exceeds maximum buffer size of "
+                + MAX_CAPACITY_BYTES
+                + " bytes: "
+                + oldSize
+                + " bytes buffered, "
+                + required
+                + " more requested");
+      }
+      ByteBuffer newBuffer = ByteBuffer.allocate((int) newSize);
       // copy over old content so it stays at the far end
-      remaining = newSize - oldSize;
+      remaining = (int) newSize - oldSize;
       newBuffer.position(remaining);
       newBuffer.put(oldBuffer);
       buffer = newBuffer;

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

@@ -15,6 +15,9 @@ public abstract class OtlpTraceCollector {
   /** Collects all spans added since the last collection. */
   public abstract OtlpPayload collectTraces();
 
+  /** Returns the number of bytes buffered since the last collection. */
+  public abstract int sizeInBytes();
+
   protected final boolean shouldExport(CoreSpan<?> span) {
     return span.samplingPriority() > 0 // trace-level sampling priority
         || span.getTag(SPAN_SAMPLING_MECHANISM_TAG) != null; // span-level sampling priority

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

@@ -2,6 +2,7 @@
 
 import static datadog.trace.core.otlp.common.OtlpCommonJson.writeScopeAndSchema;
 import static datadog.trace.core.otlp.common.OtlpPayload.JSON_CONTENT_TYPE;
+import static datadog.trace.core.otlp.common.OtlpProtoBuffer.MAX_CAPACITY_BYTES;
 import static datadog.trace.core.otlp.common.OtlpResourceJson.TRACE_RESOURCE_FRAGMENT;
 import static datadog.trace.core.otlp.trace.OtlpTraceJson.writeSpan;
 
@@ -54,11 +55,22 @@ public void addTrace(List<? extends CoreSpan<?>> spans) {
       payloadStarted = true;
     }
 
-    for (CoreSpan<?> span : spans) {
-      visitSpan(span);
+    try {
+      for (CoreSpan<?> span : spans) {
+        visitSpan(span);
+      }
+    } catch (Throwable e) {
+      // reset the buffer for subsequent traces
+      stop();
+      throw e;
     }
   }
 
+  @Override
+  public int sizeInBytes() {
+    return writer == null ? 0 : writer.sizeInBytes();
+  }
+
   /**
    * Marshals the traces collected so far into a JSON payload.
    *
@@ -178,5 +190,10 @@ private void completeSpan() {
     // reset temporary elements for next span
     currentSpan = null;
     currentSpanLinks = Collections.emptyList();
+
+    if (writer.sizeInBytes() > MAX_CAPACITY_BYTES) {
+      throw new IllegalStateException(
+          "OTLP payload exceeds maximum buffer size of " + MAX_CAPACITY_BYTES + " bytes");
+    }
   }
 }

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

@@ -56,12 +56,23 @@ public void addTrace(List<? extends CoreSpan<?>> spans) {
       payloadStarted = true;
     }
 
-    // OtlpProtoBuffer collects spans in reverse
-    for (int i = spans.size() - 1; i >= 0; i--) {
-      visitSpan(spans.get(i));
+    try {
+      // OtlpProtoBuffer collects spans in reverse
+      for (int i = spans.size() - 1; i >= 0; i--) {
+        visitSpan(spans.get(i));
+      }
+    } catch (Throwable e) {
+      // reset the buffer for subsequent traces
+      stop();
+      throw e;
     }
   }
 
+  @Override
+  public int sizeInBytes() {
+    return protobuf.sizeInBytes();
+  }
+
   /**
    * Marshals the traces collected so far into a chunked payload.
    *