Enforce OTLP request size limits

docs/apidiffs/current_vs_latest/opentelemetry-exporter-otlp.txt

@@ -1,2 +1,19 @@
 Comparing source compatibility of opentelemetry-exporter-otlp-1.63.0-SNAPSHOT.jar against opentelemetry-exporter-otlp-1.62.0.jar
-No changes.
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder setMaxRequestBodySize(long)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder setMaxRequestBodySize(long)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder setMaxRequestBodySize(long)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.logs.OtlpGrpcLogRecordExporterBuilder setMaxRequestMessageSize(long)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporterBuilder setMaxRequestMessageSize(long)
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporterBuilder setMaxRequestMessageSize(long)

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/http/logs/OtlpHttpLogRecordExporterBuilder.java

@@ -37,13 +37,15 @@ public final class OtlpHttpLogRecordExporterBuilder {
 
   private static final String DEFAULT_ENDPOINT = "http://localhost:4318/v1/logs";
   private static final MemoryMode DEFAULT_MEMORY_MODE = MemoryMode.REUSABLE_DATA;
+  private static final long DEFAULT_MAX_REQUEST_BODY_SIZE_BYTES = 64 * 1024L * 1024L;
 
   private final HttpExporterBuilder delegate;
   private MemoryMode memoryMode;
 
   OtlpHttpLogRecordExporterBuilder(HttpExporterBuilder delegate, MemoryMode memoryMode) {
     this.delegate = delegate;
     this.memoryMode = memoryMode;
+    this.delegate.setMaxRequestBodySize(DEFAULT_MAX_REQUEST_BODY_SIZE_BYTES);
     OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
@@ -98,6 +100,13 @@ public OtlpHttpLogRecordExporterBuilder setConnectTimeout(Duration timeout) {
     return this;
   }
 
+  /** Sets the maximum OTLP HTTP request body size in bytes. If unset, defaults to 64 MiB. */
+  public OtlpHttpLogRecordExporterBuilder setMaxRequestBodySize(long maxRequestBodySizeBytes) {
+    checkArgument(maxRequestBodySizeBytes >= 0, "maxRequestBodySizeBytes must be non-negative");
+    delegate.setMaxRequestBodySize(maxRequestBodySizeBytes);
+    return this;
+  }
+
   /**
    * Sets the OTLP endpoint to connect to. If unset, defaults to {@value DEFAULT_ENDPOINT}. The
    * endpoint must start with either http:// or https://, and include the full HTTP path.

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilder.java

@@ -46,6 +46,7 @@ public final class OtlpHttpMetricExporterBuilder {
   private static final AggregationTemporalitySelector DEFAULT_AGGREGATION_TEMPORALITY_SELECTOR =
       AggregationTemporalitySelector.alwaysCumulative();
   private static final MemoryMode DEFAULT_MEMORY_MODE = MemoryMode.REUSABLE_DATA;
+  private static final long DEFAULT_MAX_REQUEST_BODY_SIZE_BYTES = 64 * 1024L * 1024L;
 
   private final HttpExporterBuilder delegate;
 
@@ -62,6 +63,7 @@ public final class OtlpHttpMetricExporterBuilder {
     this.aggregationTemporalitySelector = aggregationTemporalitySelector;
     this.defaultAggregationSelector = defaultAggregationSelector;
     this.memoryMode = memoryMode;
+    this.delegate.setMaxRequestBodySize(DEFAULT_MAX_REQUEST_BODY_SIZE_BYTES);
     OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
@@ -118,6 +120,13 @@ public OtlpHttpMetricExporterBuilder setConnectTimeout(Duration timeout) {
     return this;
   }
 
+  /** Sets the maximum OTLP HTTP request body size in bytes. If unset, defaults to 64 MiB. */
+  public OtlpHttpMetricExporterBuilder setMaxRequestBodySize(long maxRequestBodySizeBytes) {
+    checkArgument(maxRequestBodySizeBytes >= 0, "maxRequestBodySizeBytes must be non-negative");
+    delegate.setMaxRequestBodySize(maxRequestBodySizeBytes);
+    return this;
+  }
+
   /**
    * Sets the OTLP endpoint to connect to. If unset, defaults to {@value DEFAULT_ENDPOINT}. The
    * endpoint must start with either http:// or https://, and include the full HTTP path.

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/http/trace/OtlpHttpSpanExporterBuilder.java

@@ -37,13 +37,15 @@ public final class OtlpHttpSpanExporterBuilder {
 
   private static final String DEFAULT_ENDPOINT = "http://localhost:4318/v1/traces";
   private static final MemoryMode DEFAULT_MEMORY_MODE = MemoryMode.REUSABLE_DATA;
+  private static final long DEFAULT_MAX_REQUEST_BODY_SIZE_BYTES = 64 * 1024L * 1024L;
 
   private final HttpExporterBuilder delegate;
   private MemoryMode memoryMode;
 
   OtlpHttpSpanExporterBuilder(HttpExporterBuilder delegate, MemoryMode memoryMode) {
     this.delegate = delegate;
     this.memoryMode = memoryMode;
+    this.delegate.setMaxRequestBodySize(DEFAULT_MAX_REQUEST_BODY_SIZE_BYTES);
     OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
@@ -98,6 +100,13 @@ public OtlpHttpSpanExporterBuilder setConnectTimeout(Duration timeout) {
     return this;
   }
 
+  /** Sets the maximum OTLP HTTP request body size in bytes. If unset, defaults to 64 MiB. */
+  public OtlpHttpSpanExporterBuilder setMaxRequestBodySize(long maxRequestBodySizeBytes) {
+    checkArgument(maxRequestBodySizeBytes >= 0, "maxRequestBodySizeBytes must be non-negative");
+    delegate.setMaxRequestBodySize(maxRequestBodySizeBytes);
+    return this;
+  }
+
   /**
    * Sets the OTLP endpoint to connect to. If unset, defaults to {@value DEFAULT_ENDPOINT}. The
    * endpoint must start with either http:// or https://, and include the full HTTP path.

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/internal/GrpcExporter.java

@@ -14,8 +14,11 @@
 import io.opentelemetry.sdk.common.export.GrpcResponse;
 import io.opentelemetry.sdk.common.export.GrpcSender;
 import io.opentelemetry.sdk.common.export.GrpcStatusCode;
+import io.opentelemetry.sdk.common.export.MessageWriter;
 import io.opentelemetry.sdk.common.internal.StandardComponentId;
 import io.opentelemetry.sdk.common.internal.ThrottlingLogger;
+import java.io.IOException;
+import java.io.OutputStream;
 import java.net.URI;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Supplier;
@@ -42,18 +45,36 @@ public final class GrpcExporter {
   private final String type;
   private final GrpcSender grpcSender;
   private final ExporterInstrumentation exporterMetrics;
+  private final long maxRequestMessageSize;
 
   public GrpcExporter(
       GrpcSender grpcSender,
       InternalTelemetryVersion internalTelemetryVersion,
       StandardComponentId componentId,
       Supplier<MeterProvider> meterProviderSupplier,
       URI endpoint) {
+    this(
+        grpcSender,
+        internalTelemetryVersion,
+        componentId,
+        meterProviderSupplier,
+        endpoint,
+        Long.MAX_VALUE);
+  }
+
+  public GrpcExporter(
+      GrpcSender grpcSender,
+      InternalTelemetryVersion internalTelemetryVersion,
+      StandardComponentId componentId,
+      Supplier<MeterProvider> meterProviderSupplier,
+      URI endpoint,
+      long maxRequestMessageSize) {
     this.type = componentId.getStandardType().signal().logFriendlyName();
     this.grpcSender = grpcSender;
     this.exporterMetrics =
         new ExporterInstrumentation(
             internalTelemetryVersion, meterProviderSupplier, componentId, endpoint);
+    this.maxRequestMessageSize = maxRequestMessageSize;
   }
 
   public CompletableResultCode export(Marshaler exportRequest, int numItems) {
@@ -65,15 +86,70 @@ public CompletableResultCode export(Marshaler exportRequest, int numItems) {
         exporterMetrics.startRecordingExport(numItems);
 
     CompletableResultCode result = new CompletableResultCode();
+    MessageWriter messageWriter = exportRequest.toBinaryMessageWriter();
+
+    long requestMessageSize = getRequestMessageSize(messageWriter);
+    if (requestMessageSize > maxRequestMessageSize) {
+      return failRequestTooLarge(result, metricRecording, requestMessageSize);
+    }
 
     grpcSender.send(
-        exportRequest.toBinaryMessageWriter(),
+        messageWriter,
         grpcResponse -> onResponse(result, metricRecording, grpcResponse),
         throwable -> onError(result, metricRecording, throwable));
 
     return result;
   }
 
+  private CompletableResultCode failRequestTooLarge(
+      CompletableResultCode result,
+      ExporterInstrumentation.Recording metricRecording,
+      long requestMessageSize) {
+    String errorMessage =
+        "OTLP gRPC request message size "
+            + requestMessageSize
+            + " exceeded limit of "
+            + maxRequestMessageSize
+            + " bytes";
+    IOException exception = new IOException(errorMessage);
+    metricRecording.finishFailed(exception);
+    logger.log(Level.WARNING, errorMessage);
+    result.failExceptionally(FailedExportException.grpcFailedExceptionally(exception));
+    return result;
+  }
+
+  private static long getRequestMessageSize(MessageWriter messageWriter) {
+    int contentLength = messageWriter.getContentLength();
+    if (contentLength >= 0) {
+      return contentLength;
+    }
+    try {
+      CountingOutputStream countingOutputStream = new CountingOutputStream();
+      messageWriter.writeMessage(countingOutputStream);
+      return countingOutputStream.getCount();
+    } catch (IOException e) {
+      return Long.MAX_VALUE;
+    }
+  }
+
+  private static final class CountingOutputStream extends OutputStream {
+    private long count;
+
+    @Override
+    public void write(int b) {
+      count++;
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) {
+      count += len;
+    }
+
+    private long getCount() {
+      return count;
+    }
+  }
+
   private void onResponse(
       CompletableResultCode result,
       ExporterInstrumentation.Recording metricRecording,

exporters/otlp/all/src/main/java/io/opentelemetry/exporter/otlp/internal/GrpcExporterBuilder.java

@@ -47,6 +47,7 @@
 public class GrpcExporterBuilder {
 
   public static final long DEFAULT_CONNECT_TIMEOUT_SECS = 10;
+  public static final long DEFAULT_MAX_REQUEST_MESSAGE_SIZE = Long.MAX_VALUE;
 
   private static final Logger LOGGER = Logger.getLogger(GrpcExporterBuilder.class.getName());
 
@@ -68,6 +69,7 @@ public class GrpcExporterBuilder {
   private ComponentLoader componentLoader =
       ComponentLoader.forClassLoader(GrpcExporterBuilder.class.getClassLoader());
   @Nullable private ExecutorService executorService;
+  private long maxRequestMessageSize = DEFAULT_MAX_REQUEST_MESSAGE_SIZE;
 
   // Use Object type since gRPC may not be on the classpath.
   @Nullable private Object grpcChannel;
@@ -170,6 +172,11 @@ public GrpcExporterBuilder setExecutorService(ExecutorService executorService) {
     return this;
   }
 
+  public GrpcExporterBuilder setMaxRequestMessageSize(long maxRequestMessageSize) {
+    this.maxRequestMessageSize = maxRequestMessageSize;
+    return this;
+  }
+
   @SuppressWarnings("BuilderReturnThis")
   public GrpcExporterBuilder copy() {
     GrpcExporterBuilder copy =
@@ -189,6 +196,7 @@ public GrpcExporterBuilder copy() {
     copy.internalTelemetryVersion = internalTelemetryVersion;
     copy.grpcChannel = grpcChannel;
     copy.componentLoader = componentLoader;
+    copy.maxRequestMessageSize = maxRequestMessageSize;
     return copy;
   }
 
@@ -240,7 +248,8 @@ public GrpcExporter build() {
         internalTelemetryVersion,
         ComponentId.generateLazy(exporterType),
         meterProviderSupplier,
-        endpoint);
+        endpoint,
+        maxRequestMessageSize);
   }
 
   public String toString(boolean includePrefixAndSuffix) {
@@ -272,6 +281,7 @@ public String toString(boolean includePrefixAndSuffix) {
     if (executorService != null) {
       joiner.add("executorService=" + executorService);
     }
+    joiner.add("maxRequestMessageSize=" + maxRequestMessageSize);
     joiner.add("exporterType=" + exporterType.toString());
     joiner.add("internalTelemetrySchemaVersion=" + internalTelemetryVersion);
     // Note: omit tlsConfigHelper because we can't log the configuration in any readable way