Add Compressor SPI support to OtlpGrpc{Signal}Exporters (#6103)

exporters/common/src/main/java/io/opentelemetry/exporter/internal/compression/CompressorUtil.java

@@ -5,11 +5,14 @@
 
 package io.opentelemetry.exporter.internal.compression;
 
-import java.util.Collections;
+import static io.opentelemetry.api.internal.Utils.checkArgument;
+import static java.util.stream.Collectors.joining;
+
 import java.util.HashMap;
 import java.util.Map;
 import java.util.ServiceLoader;
 import java.util.Set;
+import javax.annotation.Nullable;
 
 /**
  * Utilities for resolving SPI {@link Compressor}s.
@@ -25,23 +28,20 @@ public final class CompressorUtil {
 
   private CompressorUtil() {}
 
-  /** Get list of loaded compressors, named according to {@link Compressor#getEncoding()}. */
-  public static Set<String> supportedCompressors() {
-    return Collections.unmodifiableSet(compressorRegistry.keySet());
-  }
-
   /**
-   * Resolve the {@link Compressor} with the {@link Compressor#getEncoding()} equal to the {@code
-   * encoding}.
+   * Validate that the {@code compressionMethod} is "none" or matches a registered compressor.
    *
+   * @return {@code null} if {@code compressionMethod} is "none" or the registered compressor
    * @throws IllegalArgumentException if no match is found
    */
-  public static Compressor resolveCompressor(String encoding) {
-    Compressor compressor = compressorRegistry.get(encoding);
-    if (compressor == null) {
-      throw new IllegalArgumentException(
-          "Could not resolve compressor for encoding \"" + encoding + "\".");
-    }
+  @Nullable
+  public static Compressor validateAndResolveCompressor(String compressionMethod) {
+    Set<String> supportedEncodings = compressorRegistry.keySet();
+    Compressor compressor = compressorRegistry.get(compressionMethod);
+    checkArgument(
+        "none".equals(compressionMethod) || compressor != null,
+        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
+            + supportedEncodings.stream().collect(joining(",", "[", "]")));
     return compressor;
   }
 

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/main/resources/META-INF/native-image/io.opentelemetry.opentelemetry-exporter-common/reflect-config.json

@@ -6,5 +6,9 @@
   {
     "name":"io.opentelemetry.sdk.common.export.RetryPolicy",
     "queryAllDeclaredMethods":true
+  },
+  {
+    "name":"io.opentelemetry.exporter.internal.compression.Compressor",
+    "queryAllDeclaredMethods":true
   }
 ]

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

@@ -7,18 +7,18 @@
 
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 import static java.util.Objects.requireNonNull;
-import static java.util.stream.Collectors.joining;
 
 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;
 import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 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;
@@ -99,21 +99,14 @@ 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");
-    if (compressionMethod.equals("none")) {
-      delegate.setCompression(null);
-      return this;
-    }
-    Set<String> supportedCompressionMethods = CompressorUtil.supportedCompressors();
-    checkArgument(
-        supportedCompressionMethods.contains(compressionMethod),
-        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
-            + supportedCompressionMethods.stream().collect(joining(",", "[", "]")));
-    delegate.setCompression(CompressorUtil.resolveCompressor(compressionMethod));
+    Compressor compressor = CompressorUtil.validateAndResolveCompressor(compressionMethod);
+    delegate.setCompression(compressor);
     return this;
   }
 

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

@@ -7,9 +7,10 @@
 
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 import static java.util.Objects.requireNonNull;
-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;
@@ -21,7 +22,6 @@
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 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;
@@ -111,21 +111,14 @@ 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");
-    if (compressionMethod.equals("none")) {
-      delegate.setCompression(null);
-      return this;
-    }
-    Set<String> supportedCompressionMethods = CompressorUtil.supportedCompressors();
-    checkArgument(
-        supportedCompressionMethods.contains(compressionMethod),
-        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
-            + supportedCompressionMethods.stream().collect(joining(",", "[", "]")));
-    delegate.setCompression(CompressorUtil.resolveCompressor(compressionMethod));
+    Compressor compressor = CompressorUtil.validateAndResolveCompressor(compressionMethod);
+    delegate.setCompression(compressor);
     return this;
   }
 

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

@@ -7,18 +7,18 @@
 
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 import static java.util.Objects.requireNonNull;
-import static java.util.stream.Collectors.joining;
 
 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;
 import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 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;
@@ -99,21 +99,14 @@ 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");
-    if (compressionMethod.equals("none")) {
-      delegate.setCompression(null);
-      return this;
-    }
-    Set<String> supportedCompressionMethods = CompressorUtil.supportedCompressors();
-    checkArgument(
-        supportedCompressionMethods.contains(compressionMethod),
-        "Unsupported compressionMethod. Compression method must be \"none\" or one of: "
-            + supportedCompressionMethods.stream().collect(joining(",", "[", "]")));
-    delegate.setCompression(CompressorUtil.resolveCompressor(compressionMethod));
+    Compressor compressor = CompressorUtil.validateAndResolveCompressor(compressionMethod);
+    delegate.setCompression(compressor);
     return this;
   }
 

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

@@ -11,6 +11,9 @@
 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;
@@ -109,15 +112,14 @@ 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");
-    checkArgument(
-        compressionMethod.equals("gzip") || compressionMethod.equals("none"),
-        "Unsupported compression method. Supported compression methods include: gzip, none.");
-    delegate.setCompression(compressionMethod);
+    Compressor compressor = CompressorUtil.validateAndResolveCompressor(compressionMethod);
+    delegate.setCompression(compressor);
     return this;
   }
 

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

@@ -10,6 +10,9 @@
 
 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;
@@ -121,15 +124,14 @@ 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");
-    checkArgument(
-        compressionMethod.equals("gzip") || compressionMethod.equals("none"),
-        "Unsupported compression method. Supported compression methods include: gzip, none.");
-    delegate.setCompression(compressionMethod);
+    Compressor compressor = CompressorUtil.validateAndResolveCompressor(compressionMethod);
+    delegate.setCompression(compressor);
     return this;
   }