Add HttpSender abstraction (#5505)

exporters/common/src/main/java/io/opentelemetry/exporter/internal/auth/Authenticator.java

@@ -6,7 +6,7 @@
 package io.opentelemetry.exporter.internal.auth;
 
 import io.opentelemetry.exporter.internal.grpc.GrpcExporterBuilder;
-import io.opentelemetry.exporter.internal.okhttp.OkHttpExporterBuilder;
+import io.opentelemetry.exporter.internal.http.HttpExporterBuilder;
 import java.lang.reflect.Field;
 import java.util.Map;
 
@@ -27,7 +27,7 @@ public interface Authenticator {
   Map<String, String> getHeaders();
 
   /**
-   * Reflectively access a {@link GrpcExporterBuilder}, or {@link OkHttpExporterBuilder} instance in
+   * Reflectively access a {@link GrpcExporterBuilder}, or {@link HttpExporterBuilder} instance in
    * field called "delegate" of the instance, and set the {@link Authenticator}.
    *
    * @param builder export builder to modify
@@ -42,8 +42,8 @@ static void setAuthenticatorOnDelegate(Object builder, Authenticator authenticat
       Object value = field.get(builder);
       if (value instanceof GrpcExporterBuilder) {
         throw new IllegalArgumentException("GrpcExporterBuilder not supported yet.");
-      } else if (value instanceof OkHttpExporterBuilder) {
-        ((OkHttpExporterBuilder<?>) value).setAuthenticator(authenticator);
+      } else if (value instanceof HttpExporterBuilder) {
+        ((HttpExporterBuilder<?>) value).setAuthenticator(authenticator);
       } else {
         throw new IllegalArgumentException(
             "Delegate field is not type DefaultGrpcExporterBuilder or OkHttpGrpcExporterBuilder.");

exporters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpExporter.java

@@ -0,0 +1,145 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.exporter.internal.http;
+
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.ExporterMetrics;
+import io.opentelemetry.exporter.internal.grpc.GrpcStatusUtil;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.internal.ThrottlingLogger;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+import java.util.function.Supplier;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import javax.annotation.Nullable;
+
+/**
+ * An exporter for http/protobuf or http/json using a signal-specific Marshaler.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+@SuppressWarnings("checkstyle:JavadocMethod")
+public final class HttpExporter<T extends Marshaler> {
+
+  private static final Logger internalLogger = Logger.getLogger(HttpExporter.class.getName());
+
+  private final ThrottlingLogger logger = new ThrottlingLogger(internalLogger);
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
+
+  private final String type;
+  private final HttpSender httpSender;
+  private final ExporterMetrics exporterMetrics;
+  private final boolean exportAsJson;
+
+  public HttpExporter(
+      String exporterName,
+      String type,
+      HttpSender httpSender,
+      Supplier<MeterProvider> meterProviderSupplier,
+      boolean exportAsJson) {
+    this.type = type;
+    this.httpSender = httpSender;
+    this.exporterMetrics =
+        exportAsJson
+            ? ExporterMetrics.createHttpJson(exporterName, type, meterProviderSupplier)
+            : ExporterMetrics.createHttpProtobuf(exporterName, type, meterProviderSupplier);
+    this.exportAsJson = exportAsJson;
+  }
+
+  public CompletableResultCode export(T exportRequest, int numItems) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+
+    exporterMetrics.addSeen(numItems);
+
+    CompletableResultCode result = new CompletableResultCode();
+
+    Consumer<OutputStream> marshaler =
+        os -> {
+          try {
+            if (exportAsJson) {
+              exportRequest.writeJsonTo(os);
+            } else {
+              exportRequest.writeBinaryTo(os);
+            }
+          } catch (IOException e) {
+            throw new IllegalStateException(e);
+          }
+        };
+
+    httpSender.send(
+        marshaler,
+        exportRequest.getBinarySerializedSize(),
+        httpResponse -> {
+          int statusCode = httpResponse.statusCode();
+
+          if (statusCode >= 200 && statusCode < 300) {
+            exporterMetrics.addSuccess(numItems);
+            result.succeed();
+            return;
+          }
+
+          exporterMetrics.addFailed(numItems);
+
+          byte[] body;
+          try {
+            body = httpResponse.responseBody();
+          } catch (IOException ex) {
+            throw new IllegalStateException(ex);
+          }
+
+          String status = extractErrorStatus(httpResponse.statusMessage(), body);
+
+          logger.log(
+              Level.WARNING,
+              "Failed to export "
+                  + type
+                  + "s. Server responded with HTTP status code "
+                  + statusCode
+                  + ". Error message: "
+                  + status);
+          result.fail();
+        },
+        e -> {
+          exporterMetrics.addFailed(numItems);
+          logger.log(
+              Level.SEVERE,
+              "Failed to export "
+                  + type
+                  + "s. The request could not be executed. Full error message: "
+                  + e.getMessage(),
+              e);
+          result.fail();
+        });
+
+    return result;
+  }
+
+  public CompletableResultCode shutdown() {
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.INFO, "Calling shutdown() multiple times.");
+      return CompletableResultCode.ofSuccess();
+    }
+    return httpSender.shutdown();
+  }
+
+  private static String extractErrorStatus(String statusMessage, @Nullable byte[] responseBody) {
+    if (responseBody == null) {
+      return "Response body missing, HTTP status message: " + statusMessage;
+    }
+    try {
+      return GrpcStatusUtil.getStatusMessage(responseBody);
+    } catch (IOException e) {
+      return "Unable to parse response body, HTTP status message: " + statusMessage;
+    }
+  }
+}

exporters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpExporterBuilder.java

@@ -0,0 +1,142 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.exporter.internal.http;
+
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.metrics.MeterProvider;
+import io.opentelemetry.exporter.internal.ExporterBuilderUtil;
+import io.opentelemetry.exporter.internal.TlsConfigHelper;
+import io.opentelemetry.exporter.internal.auth.Authenticator;
+import io.opentelemetry.exporter.internal.marshal.Marshaler;
+import io.opentelemetry.exporter.internal.okhttp.OkHttpHttpSender;
+import io.opentelemetry.exporter.internal.retry.RetryPolicy;
+import java.net.URI;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
+import javax.annotation.Nullable;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.X509TrustManager;
+
+/**
+ * A builder for {@link HttpExporter}.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ */
+@SuppressWarnings("checkstyle:JavadocMethod")
+public final class HttpExporterBuilder<T extends Marshaler> {
+  public static final long DEFAULT_TIMEOUT_SECS = 10;
+
+  private final String exporterName;
+  private final String type;
+
+  private String endpoint;
+
+  private long timeoutNanos = TimeUnit.SECONDS.toNanos(DEFAULT_TIMEOUT_SECS);
+  private boolean compressionEnabled = false;
+  private boolean exportAsJson = false;
+  @Nullable private Map<String, String> headers;
+
+  private final TlsConfigHelper tlsConfigHelper = new TlsConfigHelper();
+  @Nullable private RetryPolicy retryPolicy;
+  private Supplier<MeterProvider> meterProviderSupplier = GlobalOpenTelemetry::getMeterProvider;
+  @Nullable private Authenticator authenticator;
+
+  public HttpExporterBuilder(String exporterName, String type, String defaultEndpoint) {
+    this.exporterName = exporterName;
+    this.type = type;
+
+    endpoint = defaultEndpoint;
+  }
+
+  public HttpExporterBuilder<T> setTimeout(long timeout, TimeUnit unit) {
+    timeoutNanos = unit.toNanos(timeout);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setTimeout(Duration timeout) {
+    return setTimeout(timeout.toNanos(), TimeUnit.NANOSECONDS);
+  }
+
+  public HttpExporterBuilder<T> setEndpoint(String endpoint) {
+    URI uri = ExporterBuilderUtil.validateEndpoint(endpoint);
+    this.endpoint = uri.toString();
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setCompression(String compressionMethod) {
+    this.compressionEnabled = compressionMethod.equals("gzip");
+    return this;
+  }
+
+  public HttpExporterBuilder<T> addHeader(String key, String value) {
+    if (headers == null) {
+      headers = new HashMap<>();
+    }
+    headers.put(key, value);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setAuthenticator(Authenticator authenticator) {
+    this.authenticator = authenticator;
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setTrustManagerFromCerts(byte[] trustedCertificatesPem) {
+    tlsConfigHelper.setTrustManagerFromCerts(trustedCertificatesPem);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setKeyManagerFromCerts(
+      byte[] privateKeyPem, byte[] certificatePem) {
+    tlsConfigHelper.setKeyManagerFromCerts(privateKeyPem, certificatePem);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setSslContext(
+      SSLContext sslContext, X509TrustManager trustManager) {
+    tlsConfigHelper.setSslContext(sslContext, trustManager);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setMeterProvider(MeterProvider meterProvider) {
+    this.meterProviderSupplier = () -> meterProvider;
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setRetryPolicy(RetryPolicy retryPolicy) {
+    this.retryPolicy = retryPolicy;
+    return this;
+  }
+
+  public HttpExporterBuilder<T> exportAsJson() {
+    this.exportAsJson = true;
+    return this;
+  }
+
+  public HttpExporter<T> build() {
+    Map<String, String> headers = this.headers == null ? Collections.emptyMap() : this.headers;
+    Supplier<Map<String, String>> headerSupplier = () -> headers;
+
+    HttpSender httpSender =
+        new OkHttpHttpSender(
+            endpoint,
+            compressionEnabled,
+            exportAsJson ? "application/json" : "application/x-protobuf",
+            timeoutNanos,
+            headerSupplier,
+            authenticator,
+            retryPolicy,
+            tlsConfigHelper.getSslContext(),
+            tlsConfigHelper.getTrustManager());
+
+    return new HttpExporter<>(exporterName, type, httpSender, meterProviderSupplier, exportAsJson);
+  }
+}

exporters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpSender.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.exporter.internal.http;
+
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.function.Consumer;
+
+/**
+ * An abstraction for sending HTTP requests and handling responses.
+ *
+ * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
+ * at any time.
+ *
+ * @see HttpExporter
+ * @see HttpExporterBuilder
+ */
+public interface HttpSender {
+
+  /**
+   * Send an HTTP request, including any retry attempts. {@code onResponse} is called with the HTTP
+   * response, either a success response or a error response after retries. {@code onError} is
+   * called when the request could not be executed due to cancellation, connectivity problems, or
+   * timeout.
+   *
+   * @param marshaler the request body marshaler
+   * @param contentLength the request body content length
+   * @param onResponse the callback to invoke with the HTTP response
+   * @param onError the callback to invoke when the HTTP request could not be executed
+   */
+  void send(
+      Consumer<OutputStream> marshaler,
+      int contentLength,
+      Consumer<Response> onResponse,
+      Consumer<Throwable> onError);
+
+  /** Shutdown the sender. */
+  CompletableResultCode shutdown();
+
+  /** The HTTP response. */
+  interface Response {
+
+    /** The HTTP status code. */
+    int statusCode();
+
+    /** The HTTP status message. */
+    String statusMessage();
+
+    /** The HTTP response body. */
+    byte[] responseBody() throws IOException;
+  }
+}