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.64.0-SNAPSHOT.jar against opentelemetry-exporter-otlp-1.63.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/jmh/java/io/opentelemetry/exporter/otlp/trace/OltpExporterBenchmark.java

@@ -13,6 +13,7 @@
 import io.grpc.stub.StreamObserver;
 import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.exporter.otlp.internal.GrpcExporter;
+import io.opentelemetry.exporter.otlp.internal.GrpcExporterBuilder;
 import io.opentelemetry.exporter.otlp.internal.HttpExporter;
 import io.opentelemetry.exporter.otlp.internal.HttpExporterBuilder;
 import io.opentelemetry.exporter.sender.grpc.managedchannel.internal.UpstreamGrpcSender;
@@ -95,7 +96,8 @@ public void setUp() {
             InternalTelemetryVersion.LATEST,
             ComponentId.generateLazy(StandardComponentId.ExporterType.OTLP_GRPC_SPAN_EXPORTER),
             MeterProvider::noop,
-            URI.create("http://localhost"));
+            URI.create("http://localhost"),
+            GrpcExporterBuilder.DEFAULT_MAX_REQUEST_MESSAGE_SIZE);
 
     okhttpGrpcSender =
         new GrpcExporter(
@@ -115,7 +117,8 @@ public void setUp() {
             InternalTelemetryVersion.LATEST,
             ComponentId.generateLazy(StandardComponentId.ExporterType.OTLP_GRPC_SPAN_EXPORTER),
             MeterProvider::noop,
-            URI.create("http://localhost"));
+            URI.create("http://localhost"),
+            GrpcExporterBuilder.DEFAULT_MAX_REQUEST_MESSAGE_SIZE);
 
     httpExporter =
         new HttpExporterBuilder(

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

@@ -98,6 +98,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 positive");
+    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

@@ -118,6 +118,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 positive");
+    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

@@ -98,6 +98,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 positive");
+    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,21 @@ 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) {
+      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) {
@@ -64,16 +70,73 @@ public CompletableResultCode export(Marshaler exportRequest, int numItems) {
     ExporterInstrumentation.Recording metricRecording =
         exporterMetrics.startRecordingExport(numItems);
 
+    MessageWriter messageWriter = exportRequest.toBinaryMessageWriter();
+
+    long requestMessageSize = getRequestMessageSize(messageWriter);
+    if (requestMessageSize > maxRequestMessageSize) {
+      return failRequestTooLarge(metricRecording, requestMessageSize);
+    }
+
     CompletableResultCode result = new CompletableResultCode();
 
     grpcSender.send(
-        exportRequest.toBinaryMessageWriter(),
+        messageWriter,
         grpcResponse -> onResponse(result, metricRecording, grpcResponse),
         throwable -> onError(result, metricRecording, throwable));
 
     return result;
   }
 
+  private CompletableResultCode failRequestTooLarge(
+      ExporterInstrumentation.Recording metricRecording, long requestMessageSize) {
+    String errorMessage =
+        "Failed to export "
+            + type
+            + "s. Request message size "
+            + requestMessageSize
+            + " exceeded limit of "
+            + maxRequestMessageSize
+            + " bytes";
+    IOException exception = new IOException(errorMessage);
+    metricRecording.finishFailed(exception);
+    logger.log(Level.WARNING, errorMessage);
+    CompletableResultCode result = new CompletableResultCode();
+    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 = 64 * 1024L * 1024L;
 
   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

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

@@ -42,20 +42,23 @@ public final class HttpExporter {
   private final HttpSender httpSender;
   private final ExporterInstrumentation exporterMetrics;
   private final boolean exportAsJson;
+  private final long maxRequestBodySize;
 
   public HttpExporter(
       StandardComponentId componentId,
       HttpSender httpSender,
       Supplier<MeterProvider> meterProviderSupplier,
       InternalTelemetryVersion internalTelemetryVersion,
       URI endpoint,
-      boolean exportAsJson) {
+      boolean exportAsJson,
+      long maxRequestBodySize) {
     this.type = componentId.getStandardType().signal().logFriendlyName();
     this.httpSender = httpSender;
     this.exporterMetrics =
         new ExporterInstrumentation(
             internalTelemetryVersion, meterProviderSupplier, componentId, endpoint);
     this.exportAsJson = exportAsJson;
+    this.maxRequestBodySize = maxRequestBodySize;
   }
 
   public CompletableResultCode export(Marshaler exportRequest, int numItems) {
@@ -66,10 +69,16 @@ public CompletableResultCode export(Marshaler exportRequest, int numItems) {
     ExporterInstrumentation.Recording metricRecording =
         exporterMetrics.startRecordingExport(numItems);
 
-    CompletableResultCode result = new CompletableResultCode();
     MessageWriter messageWriter =
         exportAsJson ? exportRequest.toJsonMessageWriter() : exportRequest.toBinaryMessageWriter();
 
+    int requestBodySize = messageWriter.getContentLength();
+    if (requestBodySize > maxRequestBodySize) {
+      return failRequestTooLarge(metricRecording, requestBodySize);
+    }
+
+    CompletableResultCode result = new CompletableResultCode();
+
     httpSender.send(
         messageWriter,
         httpResponse -> onResponse(result, metricRecording, httpResponse),
@@ -78,6 +87,24 @@ public CompletableResultCode export(Marshaler exportRequest, int numItems) {
     return result;
   }
 
+  private CompletableResultCode failRequestTooLarge(
+      ExporterInstrumentation.Recording metricRecording, long requestBodySize) {
+    String errorMessage =
+        "Failed to export "
+            + type
+            + "s. Request body size "
+            + requestBodySize
+            + " exceeded limit of "
+            + maxRequestBodySize
+            + " bytes";
+    IOException exception = new IOException(errorMessage);
+    metricRecording.finishFailed(exception);
+    logger.log(Level.WARNING, errorMessage);
+    CompletableResultCode result = new CompletableResultCode();
+    result.failExceptionally(FailedExportException.httpFailedExceptionally(exception));
+    return result;
+  }
+
   private void onResponse(
       CompletableResultCode result,
       ExporterInstrumentation.Recording metricRecording,

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

@@ -46,6 +46,7 @@
 public final class HttpExporterBuilder {
   public static final long DEFAULT_TIMEOUT_SECS = 10;
   public static final long DEFAULT_CONNECT_TIMEOUT_SECS = 10;
+  public static final long DEFAULT_MAX_REQUEST_BODY_SIZE = 64 * 1024L * 1024L;
 
   private static final Logger LOGGER = Logger.getLogger(HttpExporterBuilder.class.getName());
 
@@ -69,6 +70,7 @@ public final class HttpExporterBuilder {
   private ComponentLoader componentLoader =
       ComponentLoader.forClassLoader(HttpExporterBuilder.class.getClassLoader());
   @Nullable private ExecutorService executorService;
+  private long maxRequestBodySize = DEFAULT_MAX_REQUEST_BODY_SIZE;
 
   public HttpExporterBuilder(
       StandardComponentId.ExporterType exporterType, String defaultEndpoint) {
@@ -167,6 +169,11 @@ public HttpExporterBuilder setExecutorService(ExecutorService executorService) {
     return this;
   }
 
+  public HttpExporterBuilder setMaxRequestBodySize(long maxRequestBodySize) {
+    this.maxRequestBodySize = maxRequestBodySize;
+    return this;
+  }
+
   public HttpExporterBuilder exportAsJson() {
     this.exportAsJson = true;
     exporterType = mapToJsonTypeIfPossible(exporterType);
@@ -204,6 +211,7 @@ public HttpExporterBuilder copy() {
     copy.internalTelemetryVersion = internalTelemetryVersion;
     copy.proxyOptions = proxyOptions;
     copy.componentLoader = componentLoader;
+    copy.maxRequestBodySize = maxRequestBodySize;
     return copy;
   }
 
@@ -256,7 +264,8 @@ public HttpExporter build() {
         meterProviderSupplier,
         internalTelemetryVersion,
         endpoint,
-        exportAsJson);
+        exportAsJson,
+        maxRequestBodySize);
   }
 
   public String toString(boolean includePrefixAndSuffix) {
@@ -286,6 +295,7 @@ public String toString(boolean includePrefixAndSuffix) {
     if (executorService != null) {
       joiner.add("executorService=" + executorService);
     }
+    joiner.add("maxRequestBodySize=" + maxRequestBodySize);
     joiner.add("exporterType=" + exporterType);
     joiner.add("internalTelemetrySchemaVersion=" + internalTelemetryVersion);
     // Note: omit tlsConfigHelper because we can't log the configuration in any readable way

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

@@ -127,6 +127,15 @@ public OtlpGrpcLogRecordExporterBuilder setConnectTimeout(Duration timeout) {
     return this;
   }
 
+  /** Sets the maximum OTLP gRPC request message size in bytes. If unset, defaults to 64 MiB. */
+  public OtlpGrpcLogRecordExporterBuilder setMaxRequestMessageSize(
+      long maxRequestMessageSizeBytes) {
+    checkArgument(
+        maxRequestMessageSizeBytes > 0, "maxRequestMessageSizeBytes must be positive");
+    delegate.setMaxRequestMessageSize(maxRequestMessageSizeBytes);
+    return this;
+  }
+
   /**
    * Sets the OTLP endpoint to connect to. If unset, defaults to {@value DEFAULT_ENDPOINT_URL}. The
    * endpoint must start with either http:// or https://.