Fix Otlp Aws exporters failures for GZIP compressed telemetry exports (#1124)

Author: srprash

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/AwsApplicationSignalsCustomizerProvider.java

@@ -135,6 +135,12 @@ public final class AwsApplicationSignalsCustomizerProvider
   private static final String OTEL_TRACES_SAMPLER = "otel.traces.sampler";
   private static final String OTEL_TRACES_SAMPLER_ARG = "otel.traces.sampler.arg";
   static final String OTEL_EXPORTER_OTLP_LOGS_HEADERS = "otel.exporter.otlp.logs.headers";
+  private static final String OTEL_EXPORTER_OTLP_COMPRESSION_CONFIG =
+      "otel.exporter.otlp.compression";
+  private static final String OTEL_EXPORTER_OTLP_TRACES_COMPRESSION_CONFIG =
+      "otel.exporter.otlp.traces.compression";
+  private static final String OTEL_EXPORTER_OTLP_LOGS_COMPRESSION_CONFIG =
+      "otel.exporter.otlp.logs.compression";
 
   // UDP packet can be upto 64KB. To limit the packet size, we limit the exported batch size.
   // This is a bit of a magic number, as there is no simple way to tell how many spans can make a
@@ -394,11 +400,18 @@ SpanExporter customizeSpanExporter(SpanExporter spanExporter, ConfigProperties c
       // and OTEL_EXPORTER_OTLP_TRACES_PROTOCOL is http/protobuf
       // so the given spanExporter will be an instance of OtlpHttpSpanExporter
 
+      // get compression method from environment
+      String compression =
+          configProps.getString(
+              OTEL_EXPORTER_OTLP_TRACES_COMPRESSION_CONFIG,
+              configProps.getString(OTEL_EXPORTER_OTLP_COMPRESSION_CONFIG, "none"));
+
       try {
         spanExporter =
             OtlpAwsSpanExporterBuilder.create(
                     (OtlpHttpSpanExporter) spanExporter,
                     configProps.getString(OTEL_EXPORTER_OTLP_TRACES_ENDPOINT))
+                .setCompression(compression)
                 .build();
       } catch (Exception e) {
         // This technically should never happen as the validator checks for the correct env
@@ -430,10 +443,17 @@ LogRecordExporter customizeLogsExporter(
       // OTEL_EXPORTER_OTLP_LOGS_PROTOCOL is http/protobuf
       // so the given logsExporter will be an instance of OtlpHttpLogRecorderExporter
 
+      // get compression method from environment
+      String compression =
+          configProps.getString(
+              OTEL_EXPORTER_OTLP_LOGS_COMPRESSION_CONFIG,
+              configProps.getString(OTEL_EXPORTER_OTLP_COMPRESSION_CONFIG, "none"));
+
       try {
         return OtlpAwsLogsExporterBuilder.create(
                 (OtlpHttpLogRecordExporter) logsExporter,
                 configProps.getString(OTEL_EXPORTER_OTLP_LOGS_ENDPOINT))
+            .setCompression(compression)
             .build();
       } catch (Exception e) {
         // This technically should never happen as the validator checks for the correct env

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/common/SigV4AuthHeaderSupplier.java renamed to awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/common/AwsAuthHeaderSupplier.java

@@ -16,11 +16,14 @@
 package software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common;
 
 import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
 import java.net.URI;
 import java.util.*;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
+import java.util.zip.GZIPOutputStream;
 import software.amazon.awssdk.auth.credentials.AwsCredentials;
 import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
 import software.amazon.awssdk.http.SdkHttpFullRequest;
@@ -29,19 +32,19 @@
 import software.amazon.awssdk.http.auth.aws.signer.AwsV4HttpSigner;
 import software.amazon.awssdk.http.auth.spi.signer.SignedRequest;
 
-final class SigV4AuthHeaderSupplier implements Supplier<Map<String, String>> {
+final class AwsAuthHeaderSupplier implements Supplier<Map<String, String>> {
   BaseOtlpAwsExporter exporter;
   Logger logger;
 
-  public SigV4AuthHeaderSupplier(BaseOtlpAwsExporter exporter) {
+  public AwsAuthHeaderSupplier(BaseOtlpAwsExporter exporter) {
     this.exporter = exporter;
     this.logger = Logger.getLogger(exporter.getClass().getName());
   }
 
   @Override
   public Map<String, String> get() {
     try {
-      byte[] data = exporter.data.get();
+      ByteArrayOutputStream data = exporter.data.get();
 
       SdkHttpRequest httpRequest =
           SdkHttpFullRequest.builder()
@@ -50,6 +53,14 @@ public Map<String, String> get() {
               .putHeader("Content-Type", "application/x-protobuf")
               .build();
 
+      // Compress the data before signing with gzip
+      ByteArrayOutputStream compressedData;
+      if (exporter.getCompression().equals(CompressionMethod.GZIP)) {
+        compressedData = compressWithGzip(data);
+      } else {
+        compressedData = data;
+      }
+
       AwsCredentials credentials = DefaultCredentialsProvider.create().resolveCredentials();
 
       SignedRequest signedRequest =
@@ -60,7 +71,7 @@ public Map<String, String> get() {
                           .request(httpRequest)
                           .putProperty(AwsV4HttpSigner.SERVICE_SIGNING_NAME, exporter.serviceName())
                           .putProperty(AwsV4HttpSigner.REGION_NAME, exporter.awsRegion)
-                          .payload(() -> new ByteArrayInputStream(data)));
+                          .payload(() -> new ByteArrayInputStream(compressedData.toByteArray())));
 
       Map<String, String> result = new HashMap<>();
 
@@ -84,4 +95,21 @@ public Map<String, String> get() {
       return Collections.emptyMap();
     }
   }
+
+  /**
+   * Compresses the given byte array using GZIP compression.
+   *
+   * @param data the byte array stream to compress
+   * @return the compressed byte as a ByteArrayOutputStream
+   * @throws IOException if compression fails
+   */
+  private ByteArrayOutputStream compressWithGzip(ByteArrayOutputStream data) throws IOException {
+    ByteArrayOutputStream compressedData = new ByteArrayOutputStream();
+
+    try (GZIPOutputStream gzipOut = new GZIPOutputStream(compressedData)) {
+      data.writeTo(gzipOut);
+    }
+
+    return compressedData;
+  }
 }

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/common/BaseOtlpAwsExporter.java

@@ -15,6 +15,7 @@
 
 package software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common;
 
+import java.io.ByteArrayOutputStream;
 import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
@@ -27,15 +28,21 @@ public abstract class BaseOtlpAwsExporter {
 
   protected final String awsRegion;
   protected final String endpoint;
-  protected final AtomicReference<byte[]> data;
+  protected final AtomicReference<ByteArrayOutputStream> data;
   protected final Supplier<Map<String, String>> headerSupplier;
+  protected final CompressionMethod compression;
 
-  protected BaseOtlpAwsExporter(String endpoint) {
+  protected BaseOtlpAwsExporter(String endpoint, CompressionMethod compression) {
     this.endpoint = endpoint.toLowerCase();
+    this.compression = compression;
     this.awsRegion = endpoint.split("\\.")[1];
     this.data = new AtomicReference<>();
-    this.headerSupplier = new SigV4AuthHeaderSupplier(this);
+    this.headerSupplier = new AwsAuthHeaderSupplier(this);
   }
 
   public abstract String serviceName();
+
+  public CompressionMethod getCompression() {
+    return this.compression;
+  }
 }

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/common/CompressionMethod.java

@@ -0,0 +1,21 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common;
+
+public enum CompressionMethod {
+  NONE,
+  GZIP
+}

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/logs/OtlpAwsLogsExporter.java

@@ -28,6 +28,7 @@
 import java.util.StringJoiner;
 import javax.annotation.Nonnull;
 import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common.BaseOtlpAwsExporter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common.CompressionMethod;
 
 /**
  * This exporter extends the functionality of the OtlpHttpLogsRecordExporter to allow logs to be
@@ -42,19 +43,18 @@ public final class OtlpAwsLogsExporter extends BaseOtlpAwsExporter implements Lo
   private final OtlpHttpLogRecordExporter parentExporter;
 
   static OtlpAwsLogsExporter getDefault(String endpoint) {
-    return new OtlpAwsLogsExporter(endpoint);
+    return new OtlpAwsLogsExporter(
+        OtlpHttpLogRecordExporter.getDefault(), endpoint, CompressionMethod.NONE);
   }
 
-  static OtlpAwsLogsExporter create(OtlpHttpLogRecordExporter parent, String endpoint) {
-    return new OtlpAwsLogsExporter(parent, endpoint);
+  static OtlpAwsLogsExporter create(
+      OtlpHttpLogRecordExporter parent, String endpoint, CompressionMethod compression) {
+    return new OtlpAwsLogsExporter(parent, endpoint, compression);
   }
 
-  private OtlpAwsLogsExporter(String endpoint) {
-    this(OtlpHttpLogRecordExporter.getDefault(), endpoint);
-  }
-
-  private OtlpAwsLogsExporter(OtlpHttpLogRecordExporter parentExporter, String endpoint) {
-    super(endpoint);
+  private OtlpAwsLogsExporter(
+      OtlpHttpLogRecordExporter parentExporter, String endpoint, CompressionMethod compression) {
+    super(endpoint, compression);
 
     this.parentExporterBuilder =
         parentExporter.toBuilder()
@@ -75,7 +75,7 @@ public CompletableResultCode export(@Nonnull Collection<LogRecordData> logs) {
     try {
       ByteArrayOutputStream buffer = new ByteArrayOutputStream();
       LogsRequestMarshaler.create(logs).writeBinaryTo(buffer);
-      this.data.set(buffer.toByteArray());
+      this.data.set(buffer);
       return this.parentExporter.export(logs);
     } catch (IOException e) {
       return CompletableResultCode.ofFailure();

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/logs/OtlpAwsLogsExporterBuilder.java

@@ -18,10 +18,12 @@
 import static java.util.Objects.requireNonNull;
 
 import io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common.CompressionMethod;
 
 public class OtlpAwsLogsExporterBuilder {
   private final OtlpHttpLogRecordExporter parentExporter;
   private final String endpoint;
+  private String compression;
 
   public static OtlpAwsLogsExporterBuilder create(
       OtlpHttpLogRecordExporter parentExporter, String endpoint) {
@@ -32,8 +34,18 @@ public static OtlpAwsLogsExporter getDefault(String endpoint) {
     return OtlpAwsLogsExporter.getDefault(endpoint);
   }
 
+  public OtlpAwsLogsExporterBuilder setCompression(String compression) {
+    this.compression = compression;
+    return this;
+  }
+
   public OtlpAwsLogsExporter build() {
-    return OtlpAwsLogsExporter.create(this.parentExporter, this.endpoint);
+    CompressionMethod compression = CompressionMethod.NONE;
+    if (this.compression != null && "gzip".equalsIgnoreCase(this.compression)) {
+      compression = CompressionMethod.GZIP;
+    }
+
+    return OtlpAwsLogsExporter.create(this.parentExporter, this.endpoint, compression);
   }
 
   private OtlpAwsLogsExporterBuilder(OtlpHttpLogRecordExporter parentExporter, String endpoint) {

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/traces/OtlpAwsSpanExporter.java

@@ -28,6 +28,7 @@
 import java.util.StringJoiner;
 import javax.annotation.Nonnull;
 import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common.BaseOtlpAwsExporter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common.CompressionMethod;
 
 /**
  * This exporter extends the functionality of the OtlpHttpSpanExporter to allow spans to be exported
@@ -40,19 +41,18 @@ public final class OtlpAwsSpanExporter extends BaseOtlpAwsExporter implements Sp
   private final OtlpHttpSpanExporter parentExporter;
 
   static OtlpAwsSpanExporter getDefault(String endpoint) {
-    return new OtlpAwsSpanExporter(endpoint);
+    return new OtlpAwsSpanExporter(
+        OtlpHttpSpanExporter.getDefault(), endpoint, CompressionMethod.NONE);
   }
 
-  static OtlpAwsSpanExporter create(OtlpHttpSpanExporter parent, String endpoint) {
-    return new OtlpAwsSpanExporter(parent, endpoint);
+  static OtlpAwsSpanExporter create(
+      OtlpHttpSpanExporter parent, String endpoint, CompressionMethod compression) {
+    return new OtlpAwsSpanExporter(parent, endpoint, compression);
   }
 
-  private OtlpAwsSpanExporter(String endpoint) {
-    this(OtlpHttpSpanExporter.getDefault(), endpoint);
-  }
-
-  private OtlpAwsSpanExporter(OtlpHttpSpanExporter parentExporter, String endpoint) {
-    super(endpoint);
+  private OtlpAwsSpanExporter(
+      OtlpHttpSpanExporter parentExporter, String endpoint, CompressionMethod compression) {
+    super(endpoint, compression);
 
     this.parentExporterBuilder =
         parentExporter.toBuilder()
@@ -73,7 +73,7 @@ public CompletableResultCode export(@Nonnull Collection<SpanData> spans) {
     try {
       ByteArrayOutputStream buffer = new ByteArrayOutputStream();
       TraceRequestMarshaler.create(spans).writeBinaryTo(buffer);
-      this.data.set(buffer.toByteArray());
+      this.data.set(buffer);
       return this.parentExporter.export(spans);
     } catch (IOException e) {
       return CompletableResultCode.ofFailure();

awsagentprovider/src/main/java/software/amazon/opentelemetry/javaagent/providers/exporter/otlp/aws/traces/OtlpAwsSpanExporterBuilder.java

@@ -18,10 +18,12 @@
 import static java.util.Objects.requireNonNull;
 
 import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
+import software.amazon.opentelemetry.javaagent.providers.exporter.otlp.aws.common.CompressionMethod;
 
 public class OtlpAwsSpanExporterBuilder {
   private final OtlpHttpSpanExporter parentExporter;
   private final String endpoint;
+  private String compression;
 
   public static OtlpAwsSpanExporterBuilder create(
       OtlpHttpSpanExporter parentExporter, String endpoint) {
@@ -32,12 +34,22 @@ public static OtlpAwsSpanExporter getDefault(String endpoint) {
     return OtlpAwsSpanExporter.getDefault(endpoint);
   }
 
+  public OtlpAwsSpanExporterBuilder setCompression(String compression) {
+    this.compression = compression;
+    return this;
+  }
+
   private OtlpAwsSpanExporterBuilder(OtlpHttpSpanExporter parentExporter, String endpoint) {
     this.parentExporter = requireNonNull(parentExporter, "Must set a parentExporter");
     this.endpoint = requireNonNull(endpoint, "Must set an endpoint");
   }
 
   public OtlpAwsSpanExporter build() {
-    return OtlpAwsSpanExporter.create(this.parentExporter, this.endpoint);
+    CompressionMethod compression = CompressionMethod.NONE;
+    if (this.compression != null && "gzip".equalsIgnoreCase(this.compression)) {
+      compression = CompressionMethod.GZIP;
+    }
+
+    return OtlpAwsSpanExporter.create(this.parentExporter, this.endpoint, compression);
   }
 }