Add Compressor SPI support to OtlpGrpc{Signal}Exporters

exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporterBuilder.java

@@ -12,6 +12,7 @@
 import io.opentelemetry.api.metrics.MeterProvider;
 import io.opentelemetry.exporter.internal.ExporterBuilderUtil;
 import io.opentelemetry.exporter.internal.TlsConfigHelper;
+import io.opentelemetry.exporter.internal.compression.Compressor;
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
@@ -21,6 +22,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.ServiceLoader;
 import java.util.StringJoiner;
 import java.util.concurrent.TimeUnit;
@@ -51,7 +53,7 @@ public class GrpcExporterBuilder<T extends Marshaler> {
 
   private long timeoutNanos;
   private URI endpoint;
-  private boolean compressionEnabled = false;
+  @Nullable private Compressor compressor;
   private final Map<String, String> constantHeaders = new HashMap<>();
   private Supplier<Map<String, String>> headerSupplier = Collections::emptyMap;
   private TlsConfigHelper tlsConfigHelper = new TlsConfigHelper();
@@ -95,8 +97,8 @@ public GrpcExporterBuilder<T> setEndpoint(String endpoint) {
     return this;
   }
 
-  public GrpcExporterBuilder<T> setCompression(String compressionMethod) {
-    this.compressionEnabled = compressionMethod.equals("gzip");
+  public GrpcExporterBuilder<T> setCompression(@Nullable Compressor compressor) {
+    this.compressor = compressor;
     return this;
   }
 
@@ -150,7 +152,7 @@ public GrpcExporterBuilder<T> copy() {
 
     copy.timeoutNanos = timeoutNanos;
     copy.endpoint = endpoint;
-    copy.compressionEnabled = compressionEnabled;
+    copy.compressor = compressor;
     copy.constantHeaders.putAll(constantHeaders);
     copy.headerSupplier = headerSupplier;
     copy.tlsConfigHelper = tlsConfigHelper.copy();
@@ -189,7 +191,7 @@ public GrpcExporter<T> build() {
         grpcSenderProvider.createSender(
             endpoint,
             grpcEndpointPath,
-            compressionEnabled,
+            compressor,
             timeoutNanos,
             headerSupplier,
             grpcChannel,
@@ -212,7 +214,9 @@ public String toString(boolean includePrefixAndSuffix) {
     joiner.add("endpoint=" + endpoint.toString());
     joiner.add("endpointPath=" + grpcEndpointPath);
     joiner.add("timeoutNanos=" + timeoutNanos);
-    joiner.add("compressionEnabled=" + compressionEnabled);
+    joiner.add(
+        "compressorEncoding="
+            + Optional.ofNullable(compressor).map(Compressor::getEncoding).orElse(null));
     StringJoiner headersJoiner = new StringJoiner(", ", "Headers{", "}");
     constantHeaders.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
     Map<String, String> headers = headerSupplier.get();

exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcSenderProvider.java

@@ -6,6 +6,7 @@
 package io.opentelemetry.exporter.internal.grpc;
 
 import io.grpc.Channel;
+import io.opentelemetry.exporter.internal.compression.Compressor;
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
@@ -31,7 +32,7 @@ public interface GrpcSenderProvider {
   <T extends Marshaler> GrpcSender<T> createSender(
       URI endpoint,
       String endpointPath,
-      boolean compressionEnabled,
+      @Nullable Compressor compressor,
       long timeoutNanos,
       Supplier<Map<String, List<String>>> headersSupplier,
       @Nullable Object managedChannel,

exporters/common/src/test/java/io/opentelemetry/exporter/internal/grpc/GrpcExporterBuilderTest.java

@@ -7,6 +7,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import io.opentelemetry.exporter.internal.compression.GzipCompressor;
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import java.net.URI;
 import org.junit.jupiter.api.BeforeEach;
@@ -25,27 +26,27 @@ void setUp() {
 
   @Test
   void compressionDefault() {
-    assertThat(builder).extracting("compressionEnabled").isEqualTo(false);
+    assertThat(builder).extracting("compressor").isNull();
   }
 
   @Test
   void compressionNone() {
-    builder.setCompression("none");
+    builder.setCompression(null);
 
-    assertThat(builder).extracting("compressionEnabled").isEqualTo(false);
+    assertThat(builder).extracting("compressor").isNull();
   }
 
   @Test
   void compressionGzip() {
-    builder.setCompression("gzip");
+    builder.setCompression(GzipCompressor.getInstance());
 
-    assertThat(builder).extracting("compressionEnabled").isEqualTo(true);
+    assertThat(builder).extracting("compressor").isEqualTo(GzipCompressor.getInstance());
   }
 
   @Test
   void compressionEnabledAndDisabled() {
-    builder.setCompression("gzip").setCompression("none");
+    builder.setCompression(GzipCompressor.getInstance()).setCompression(null);
 
-    assertThat(builder).extracting("compressionEnabled").isEqualTo(false);
+    assertThat(builder).extracting("compressor").isNull();
   }
 }

exporters/otlp/all/src/jmh/java/io/opentelemetry/exporter/otlp/trace/OltpExporterBenchmark.java

@@ -98,7 +98,7 @@ public void setUp() {
                 URI.create("http://localhost:" + server.activeLocalPort())
                     .resolve(OtlpGrpcSpanExporterBuilder.GRPC_ENDPOINT_PATH)
                     .toString(),
-                /* compressionEnabled= */ false,
+                null,
                 10,
                 Collections::emptyMap,
                 null,

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

@@ -11,6 +11,8 @@
 
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.compression.Compressor;
+import io.opentelemetry.exporter.internal.compression.CompressorProvider;
 import io.opentelemetry.exporter.internal.compression.CompressorUtil;
 import io.opentelemetry.exporter.internal.http.HttpExporterBuilder;
 import io.opentelemetry.exporter.internal.otlp.logs.LogsRequestMarshaler;
@@ -99,8 +101,9 @@ public OtlpHttpLogRecordExporterBuilder setEndpoint(String endpoint) {
   }
 
   /**
-   * Sets the method used to compress payloads. If unset, compression is disabled. Currently
-   * supported compression methods include "gzip" and "none".
+   * Sets the method used to compress payloads. If unset, compression is disabled. Compression
+   * method "gzip" and "none" are supported out of the box. Support for additional compression
+   * methods is available by implementing {@link Compressor} and {@link CompressorProvider}.
    */
   public OtlpHttpLogRecordExporterBuilder setCompression(String compressionMethod) {
     requireNonNull(compressionMethod, "compressionMethod");

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

@@ -10,6 +10,8 @@
 import static java.util.stream.Collectors.joining;
 
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.compression.Compressor;
+import io.opentelemetry.exporter.internal.compression.CompressorProvider;
 import io.opentelemetry.exporter.internal.compression.CompressorUtil;
 import io.opentelemetry.exporter.internal.http.HttpExporterBuilder;
 import io.opentelemetry.exporter.internal.otlp.metrics.MetricsRequestMarshaler;
@@ -111,8 +113,9 @@ public OtlpHttpMetricExporterBuilder setEndpoint(String endpoint) {
   }
 
   /**
-   * Sets the method used to compress payloads. If unset, compression is disabled. Currently
-   * supported compression methods include "gzip" and "none".
+   * Sets the method used to compress payloads. If unset, compression is disabled. Compression
+   * method "gzip" and "none" are supported out of the box. Support for additional compression
+   * methods is available by implementing {@link Compressor} and {@link CompressorProvider}.
    */
   public OtlpHttpMetricExporterBuilder setCompression(String compressionMethod) {
     requireNonNull(compressionMethod, "compressionMethod");

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

@@ -11,6 +11,8 @@
 
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.compression.Compressor;
+import io.opentelemetry.exporter.internal.compression.CompressorProvider;
 import io.opentelemetry.exporter.internal.compression.CompressorUtil;
 import io.opentelemetry.exporter.internal.http.HttpExporterBuilder;
 import io.opentelemetry.exporter.internal.otlp.traces.TraceRequestMarshaler;
@@ -99,8 +101,9 @@ public OtlpHttpSpanExporterBuilder setEndpoint(String endpoint) {
   }
 
   /**
-   * Sets the method used to compress payloads. If unset, compression is disabled. Currently
-   * supported compression methods include "gzip" and "none".
+   * Sets the method used to compress payloads. If unset, compression is disabled. Compression
+   * method "gzip" and "none" are supported out of the box. Support for additional compression
+   * methods is available by implementing {@link Compressor} and {@link CompressorProvider}.
    */
   public OtlpHttpSpanExporterBuilder setCompression(String compressionMethod) {
     requireNonNull(compressionMethod, "compressionMethod");

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

@@ -7,17 +7,22 @@
 
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.joining;
 
 import io.grpc.ManagedChannel;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.compression.Compressor;
+import io.opentelemetry.exporter.internal.compression.CompressorProvider;
+import io.opentelemetry.exporter.internal.compression.CompressorUtil;
 import io.opentelemetry.exporter.internal.grpc.GrpcExporterBuilder;
 import io.opentelemetry.exporter.internal.otlp.logs.LogsRequestMarshaler;
 import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
 import java.time.Duration;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
@@ -109,15 +114,22 @@ public OtlpGrpcLogRecordExporterBuilder setEndpoint(String endpoint) {
   }
 
   /**
-   * Sets the method used to compress payloads. If unset, compression is disabled. Currently
-   * supported compression methods include "gzip" and "none".
+   * Sets the method used to compress payloads. If unset, compression is disabled. Compression
+   * method "gzip" and "none" are supported out of the box. Support for additional compression
+   * methods is available by implementing {@link Compressor} and {@link CompressorProvider}.
    */
   public OtlpGrpcLogRecordExporterBuilder setCompression(String compressionMethod) {
     requireNonNull(compressionMethod, "compressionMethod");
+    if (compressionMethod.equals("none")) {
+      delegate.setCompression(null);
+      return this;
+    }
+    Set<String> supportedCompressionMethods = CompressorUtil.supportedCompressors();
     checkArgument(
-        compressionMethod.equals("gzip") || compressionMethod.equals("none"),
-        "Unsupported compression method. Supported compression methods include: gzip, none.");
-    delegate.setCompression(compressionMethod);
+        supportedCompressionMethods.contains(compressionMethod),
+        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
+            + supportedCompressionMethods.stream().collect(joining(",", "[", "]")));
+    delegate.setCompression(CompressorUtil.resolveCompressor(compressionMethod));
     return this;
   }
 

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

@@ -7,9 +7,13 @@
 
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.joining;
 
 import io.grpc.ManagedChannel;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.compression.Compressor;
+import io.opentelemetry.exporter.internal.compression.CompressorProvider;
+import io.opentelemetry.exporter.internal.compression.CompressorUtil;
 import io.opentelemetry.exporter.internal.grpc.GrpcExporterBuilder;
 import io.opentelemetry.exporter.internal.otlp.metrics.MetricsRequestMarshaler;
 import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent;
@@ -21,6 +25,7 @@
 import java.net.URI;
 import java.time.Duration;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
@@ -121,15 +126,22 @@ public OtlpGrpcMetricExporterBuilder setEndpoint(String endpoint) {
   }
 
   /**
-   * Sets the method used to compress payloads. If unset, compression is disabled. Currently
-   * supported compression methods include "gzip" and "none".
+   * Sets the method used to compress payloads. If unset, compression is disabled. Compression
+   * method "gzip" and "none" are supported out of the box. Support for additional compression
+   * methods is available by implementing {@link Compressor} and {@link CompressorProvider}.
    */
   public OtlpGrpcMetricExporterBuilder setCompression(String compressionMethod) {
     requireNonNull(compressionMethod, "compressionMethod");
+    if (compressionMethod.equals("none")) {
+      delegate.setCompression(null);
+      return this;
+    }
+    Set<String> supportedCompressionMethods = CompressorUtil.supportedCompressors();
     checkArgument(
-        compressionMethod.equals("gzip") || compressionMethod.equals("none"),
-        "Unsupported compression method. Supported compression methods include: gzip, none.");
-    delegate.setCompression(compressionMethod);
+        supportedCompressionMethods.contains(compressionMethod),
+        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
+            + supportedCompressionMethods.stream().collect(joining(",", "[", "]")));
+    delegate.setCompression(CompressorUtil.resolveCompressor(compressionMethod));
     return this;
   }
 

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

@@ -7,17 +7,22 @@
 
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 import static java.util.Objects.requireNonNull;
+import static java.util.stream.Collectors.joining;
 
 import io.grpc.ManagedChannel;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.compression.Compressor;
+import io.opentelemetry.exporter.internal.compression.CompressorProvider;
+import io.opentelemetry.exporter.internal.compression.CompressorUtil;
 import io.opentelemetry.exporter.internal.grpc.GrpcExporterBuilder;
 import io.opentelemetry.exporter.internal.otlp.traces.TraceRequestMarshaler;
 import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
 import java.time.Duration;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
@@ -105,15 +110,22 @@ public OtlpGrpcSpanExporterBuilder setEndpoint(String endpoint) {
   }
 
   /**
-   * Sets the method used to compress payloads. If unset, compression is disabled. Currently
-   * supported compression methods include "gzip" and "none".
+   * Sets the method used to compress payloads. If unset, compression is disabled. Compression
+   * method "gzip" and "none" are supported out of the box. Support for additional compression
+   * methods is available by implementing {@link Compressor} and {@link CompressorProvider}.
    */
   public OtlpGrpcSpanExporterBuilder setCompression(String compressionMethod) {
     requireNonNull(compressionMethod, "compressionMethod");
+    if (compressionMethod.equals("none")) {
+      delegate.setCompression(null);
+      return this;
+    }
+    Set<String> supportedCompressionMethods = CompressorUtil.supportedCompressors();
     checkArgument(
-        compressionMethod.equals("gzip") || compressionMethod.equals("none"),
-        "Unsupported compression method. Supported compression methods include: gzip, none.");
-    delegate.setCompression(compressionMethod);
+        supportedCompressionMethods.contains(compressionMethod),
+        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
+            + supportedCompressionMethods.stream().collect(joining(",", "[", "]")));
+    delegate.setCompression(CompressorUtil.resolveCompressor(compressionMethod));
     return this;
   }