Add OTLP header supplier configuration option (#6004)

open-telemetry · Dec 8, 2023 · 4c60397 · 4c60397
1 parent 9f3456f
commit 4c60397
Show file tree

Hide file tree

Showing 31 changed files with 372 additions and 93 deletions.
diff --git a/docs/apidiffs/current_vs_latest/opentelemetry-exporter-otlp.txt b/docs/apidiffs/current_vs_latest/opentelemetry-exporter-otlp.txt
@@ -3,11 +3,23 @@ Comparing source compatibility of  against
 	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
 	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder setConnectTimeout(long, java.util.concurrent.TimeUnit)
 	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder setConnectTimeout(java.time.Duration)
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.lang.String>>)
 ***  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 setConnectTimeout(long, java.util.concurrent.TimeUnit)
 	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder setConnectTimeout(java.time.Duration)
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.metrics.OtlpHttpMetricExporterBuilder setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.lang.String>>)
 ***  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 setConnectTimeout(long, java.util.concurrent.TimeUnit)
 	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder setConnectTimeout(java.time.Duration)
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporterBuilder setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.lang.String>>)
+***  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 setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.lang.String>>)
+***  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 setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.lang.String>>)
+***  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 setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.lang.String>>)
diff --git a/...ers/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporterBuilder.java b/...ers/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporterBuilder.java
@@ -16,7 +16,10 @@
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
 import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
 import java.util.StringJoiner;
@@ -49,7 +52,8 @@ public class GrpcExporterBuilder<T extends Marshaler> {
   private long timeoutNanos;
   private URI endpoint;
   private boolean compressionEnabled = false;
-  private final Map<String, String> headers = new HashMap<>();
+  private final Map<String, String> constantHeaders = new HashMap<>();
+  private Supplier<Map<String, String>> headerSupplier = Collections::emptyMap;
   private TlsConfigHelper tlsConfigHelper = new TlsConfigHelper();
   @Nullable private RetryPolicy retryPolicy;
   private Supplier<MeterProvider> meterProviderSupplier = GlobalOpenTelemetry::getMeterProvider;
@@ -113,8 +117,13 @@ public GrpcExporterBuilder<T> setSslContext(
     return this;
   }
 
-  public GrpcExporterBuilder<T> addHeader(String key, String value) {
-    headers.put(key, value);
+  public GrpcExporterBuilder<T> addConstantHeader(String key, String value) {
+    constantHeaders.put(key, value);
+    return this;
+  }
+
+  public GrpcExporterBuilder<T> setHeadersSupplier(Supplier<Map<String, String>> headerSupplier) {
+    this.headerSupplier = headerSupplier;
     return this;
   }
 
@@ -142,7 +151,8 @@ public GrpcExporterBuilder<T> copy() {
     copy.timeoutNanos = timeoutNanos;
     copy.endpoint = endpoint;
     copy.compressionEnabled = compressionEnabled;
-    copy.headers.putAll(headers);
+    copy.constantHeaders.putAll(constantHeaders);
+    copy.headerSupplier = headerSupplier;
     copy.tlsConfigHelper = tlsConfigHelper.copy();
     if (retryPolicy != null) {
       copy.retryPolicy = retryPolicy.toBuilder().build();
@@ -153,14 +163,35 @@ public GrpcExporterBuilder<T> copy() {
   }
 
   public GrpcExporter<T> build() {
+    Supplier<Map<String, List<String>>> headerSupplier =
+        () -> {
+          Map<String, List<String>> result = new HashMap<>();
+          Map<String, String> supplierResult = this.headerSupplier.get();
+          if (supplierResult != null) {
+            supplierResult.forEach(
+                (key, value) -> result.put(key, Collections.singletonList(value)));
+          }
+          constantHeaders.forEach(
+              (key, value) ->
+                  result.merge(
+                      key,
+                      Collections.singletonList(value),
+                      (v1, v2) -> {
+                        List<String> merged = new ArrayList<>(v1);
+                        merged.addAll(v2);
+                        return merged;
+                      }));
+          return result;
+        };
+
     GrpcSenderProvider grpcSenderProvider = resolveGrpcSenderProvider();
     GrpcSender<T> grpcSender =
         grpcSenderProvider.createSender(
             endpoint,
             grpcEndpointPath,
             compressionEnabled,
             timeoutNanos,
-            headers,
+            headerSupplier,
             grpcChannel,
             grpcStubFactory,
             retryPolicy,
@@ -183,7 +214,11 @@ public String toString(boolean includePrefixAndSuffix) {
     joiner.add("timeoutNanos=" + timeoutNanos);
     joiner.add("compressionEnabled=" + compressionEnabled);
     StringJoiner headersJoiner = new StringJoiner(", ", "Headers{", "}");
-    headers.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
+    constantHeaders.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
+    Map<String, String> headers = headerSupplier.get();
+    if (headers != null) {
+      headers.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
+    }
     joiner.add("headers=" + headersJoiner);
     if (retryPolicy != null) {
       joiner.add("retryPolicy=" + retryPolicy);

diff --git a/...ters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcSenderProvider.java b/...ters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcSenderProvider.java
@@ -9,6 +9,7 @@
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
+import java.util.List;
 import java.util.Map;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
@@ -32,7 +33,7 @@ <T extends Marshaler> GrpcSender<T> createSender(
       String endpointPath,
       boolean compressionEnabled,
       long timeoutNanos,
-      Map<String, String> headers,
+      Supplier<Map<String, List<String>>> headersSupplier,
       @Nullable Object managedChannel,
       Supplier<BiFunction<Channel, String, MarshalerServiceStub<T, ?, ?>>> stubFactory,
       @Nullable RetryPolicy retryPolicy,

diff --git a/...ers/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpExporterBuilder.java b/...ers/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpExporterBuilder.java
@@ -14,8 +14,10 @@
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.ServiceLoader;
 import java.util.StringJoiner;
@@ -49,7 +51,8 @@ public final class HttpExporterBuilder<T extends Marshaler> {
   private long connectTimeoutNanos = TimeUnit.SECONDS.toNanos(DEFAULT_CONNECT_TIMEOUT_SECS);
   private boolean compressionEnabled = false;
   private boolean exportAsJson = false;
-  @Nullable private Map<String, String> headers;
+  private final Map<String, String> constantHeaders = new HashMap<>();
+  private Supplier<Map<String, String>> headerSupplier = Collections::emptyMap;
 
   private TlsConfigHelper tlsConfigHelper = new TlsConfigHelper();
   @Nullable private RetryPolicy retryPolicy;
@@ -84,11 +87,13 @@ public HttpExporterBuilder<T> setCompression(String compressionMethod) {
     return this;
   }
 
-  public HttpExporterBuilder<T> addHeader(String key, String value) {
-    if (headers == null) {
-      headers = new HashMap<>();
-    }
-    headers.put(key, value);
+  public HttpExporterBuilder<T> addConstantHeaders(String key, String value) {
+    constantHeaders.put(key, value);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setHeadersSupplier(Supplier<Map<String, String>> headerSupplier) {
+    this.headerSupplier = headerSupplier;
     return this;
   }
 
@@ -137,9 +142,8 @@ public HttpExporterBuilder<T> copy() {
     copy.connectTimeoutNanos = connectTimeoutNanos;
     copy.exportAsJson = exportAsJson;
     copy.compressionEnabled = compressionEnabled;
-    if (headers != null) {
-      copy.headers = new HashMap<>(headers);
-    }
+    copy.constantHeaders.putAll(constantHeaders);
+    copy.headerSupplier = headerSupplier;
     copy.tlsConfigHelper = tlsConfigHelper.copy();
     if (retryPolicy != null) {
       copy.retryPolicy = retryPolicy.toBuilder().build();
@@ -150,8 +154,26 @@ public HttpExporterBuilder<T> copy() {
   }
 
   public HttpExporter<T> build() {
-    Map<String, String> headers = this.headers == null ? Collections.emptyMap() : this.headers;
-    Supplier<Map<String, String>> headerSupplier = () -> headers;
+    Supplier<Map<String, List<String>>> headerSupplier =
+        () -> {
+          Map<String, List<String>> result = new HashMap<>();
+          Map<String, String> supplierResult = this.headerSupplier.get();
+          if (supplierResult != null) {
+            supplierResult.forEach(
+                (key, value) -> result.put(key, Collections.singletonList(value)));
+          }
+          constantHeaders.forEach(
+              (key, value) ->
+                  result.merge(
+                      key,
+                      Collections.singletonList(value),
+                      (v1, v2) -> {
+                        List<String> merged = new ArrayList<>(v1);
+                        merged.addAll(v2);
+                        return merged;
+                      }));
+          return result;
+        };
 
     HttpSenderProvider httpSenderProvider = resolveHttpSenderProvider();
     HttpSender httpSender =
@@ -183,11 +205,13 @@ public String toString(boolean includePrefixAndSuffix) {
     joiner.add("connectTimeoutNanos=" + connectTimeoutNanos);
     joiner.add("compressionEnabled=" + compressionEnabled);
     joiner.add("exportAsJson=" + exportAsJson);
+    StringJoiner headersJoiner = new StringJoiner(", ", "Headers{", "}");
+    constantHeaders.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
+    Map<String, String> headers = headerSupplier.get();
     if (headers != null) {
-      StringJoiner headersJoiner = new StringJoiner(", ", "Headers{", "}");
       headers.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
-      joiner.add("headers=" + headersJoiner);
     }
+    joiner.add("headers=" + headersJoiner);
     if (retryPolicy != null) {
       joiner.add("retryPolicy=" + retryPolicy);
     }

diff --git a/...ters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpSenderProvider.java b/...ters/common/src/main/java/io/opentelemetry/exporter/internal/http/HttpSenderProvider.java
@@ -7,6 +7,7 @@
 
 import io.opentelemetry.exporter.internal.auth.Authenticator;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
 import javax.annotation.Nullable;
@@ -30,7 +31,7 @@ HttpSender createSender(
       String contentType,
       long timeoutNanos,
       long connectTimeout,
-      Supplier<Map<String, String>> headerSupplier,
+      Supplier<Map<String, List<String>>> headerSupplier,
       @Nullable Authenticator authenticator,
       @Nullable RetryPolicy retryPolicy,
       @Nullable SSLContext sslContext,

diff --git a/...ers/otlp/all/src/jmh/java/io/opentelemetry/exporter/otlp/trace/OltpExporterBenchmark.java b/...ers/otlp/all/src/jmh/java/io/opentelemetry/exporter/otlp/trace/OltpExporterBenchmark.java
@@ -84,7 +84,9 @@ public void setUp() {
             "otlp",
             "span",
             new UpstreamGrpcSender<>(
-                MarshalerTraceServiceGrpc.newFutureStub(defaultGrpcChannel, null), 10),
+                MarshalerTraceServiceGrpc.newFutureStub(defaultGrpcChannel, null),
+                10,
+                Collections::emptyMap),
             MeterProvider::noop);
 
     okhttpGrpcSender =
@@ -97,7 +99,7 @@ public void setUp() {
                     .toString(),
                 /* compressionEnabled= */ false,
                 10,
-                Collections.emptyMap(),
+                Collections::emptyMap,
                 null,
                 null,
                 null),

diff --git a/.../main/java/io/opentelemetry/exporter/otlp/http/logs/OtlpHttpLogRecordExporterBuilder.java b/.../main/java/io/opentelemetry/exporter/otlp/http/logs/OtlpHttpLogRecordExporterBuilder.java
@@ -15,6 +15,7 @@
 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.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
@@ -33,7 +34,7 @@ public final class OtlpHttpLogRecordExporterBuilder {
 
   OtlpHttpLogRecordExporterBuilder(HttpExporterBuilder<LogsRequestMarshaler> delegate) {
     this.delegate = delegate;
-    OtlpUserAgent.addUserAgentHeader(delegate::addHeader);
+    OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
   OtlpHttpLogRecordExporterBuilder() {
@@ -103,9 +104,21 @@ public OtlpHttpLogRecordExporterBuilder setCompression(String compressionMethod)
     return this;
   }
 
-  /** Add header to requests. */
+  /**
+   * Add a constant header to requests. If the {@code key} collides with another constant header
+   * name or a one from {@link #setHeaders(Supplier)}, the values from both are included.
+   */
   public OtlpHttpLogRecordExporterBuilder addHeader(String key, String value) {
-    delegate.addHeader(key, value);
+    delegate.addConstantHeaders(key, value);
+    return this;
+  }
+
+  /**
+   * Set the supplier of headers to add to requests. If a key from the map collides with a constant
+   * from {@link #addHeader(String, String)}, the values from both are included.
+   */
+  public OtlpHttpLogRecordExporterBuilder setHeaders(Supplier<Map<String, String>> headerSupplier) {
+    delegate.setHeadersSupplier(headerSupplier);
     return this;
   }
 

diff --git a/.../main/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilder.java b/.../main/java/io/opentelemetry/exporter/otlp/http/metrics/OtlpHttpMetricExporterBuilder.java
@@ -18,7 +18,9 @@
 import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.time.Duration;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
 import javax.net.ssl.X509TrustManager;
 
@@ -44,7 +46,7 @@ public final class OtlpHttpMetricExporterBuilder {
   OtlpHttpMetricExporterBuilder(HttpExporterBuilder<MetricsRequestMarshaler> delegate) {
     this.delegate = delegate;
     delegate.setMeterProvider(MeterProvider::noop);
-    OtlpUserAgent.addUserAgentHeader(delegate::addHeader);
+    OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
   OtlpHttpMetricExporterBuilder() {
@@ -114,9 +116,21 @@ public OtlpHttpMetricExporterBuilder setCompression(String compressionMethod) {
     return this;
   }
 
-  /** Add header to requests. */
+  /**
+   * Add a constant header to requests. If the {@code key} collides with another constant header
+   * name or a one from {@link #setHeaders(Supplier)}, the values from both are included.
+   */
   public OtlpHttpMetricExporterBuilder addHeader(String key, String value) {
-    delegate.addHeader(key, value);
+    delegate.addConstantHeaders(key, value);
+    return this;
+  }
+
+  /**
+   * Set the supplier of headers to add to requests. If a key from the map collides with a constant
+   * from {@link #addHeader(String, String)}, the values from both are included.
+   */
+  public OtlpHttpMetricExporterBuilder setHeaders(Supplier<Map<String, String>> headerSupplier) {
+    delegate.setHeadersSupplier(headerSupplier);
     return this;
   }
 

diff --git a/.../src/main/java/io/opentelemetry/exporter/otlp/http/trace/OtlpHttpSpanExporterBuilder.java b/.../src/main/java/io/opentelemetry/exporter/otlp/http/trace/OtlpHttpSpanExporterBuilder.java
@@ -15,6 +15,7 @@
 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.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
@@ -33,7 +34,7 @@ public final class OtlpHttpSpanExporterBuilder {
 
   OtlpHttpSpanExporterBuilder(HttpExporterBuilder<TraceRequestMarshaler> delegate) {
     this.delegate = delegate;
-    OtlpUserAgent.addUserAgentHeader(delegate::addHeader);
+    OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
   OtlpHttpSpanExporterBuilder() {
@@ -103,9 +104,21 @@ public OtlpHttpSpanExporterBuilder setCompression(String compressionMethod) {
     return this;
   }
 
-  /** Add header to requests. */
+  /**
+   * Add a constant header to requests. If the {@code key} collides with another constant header
+   * name or a one from {@link #setHeaders(Supplier)}, the values from both are included.
+   */
   public OtlpHttpSpanExporterBuilder addHeader(String key, String value) {
-    delegate.addHeader(key, value);
+    delegate.addConstantHeaders(key, value);
+    return this;
+  }
+
+  /**
+   * Set the supplier of headers to add to requests. If a key from the map collides with a constant
+   * from {@link #addHeader(String, String)}, the values from both are included.
+   */
+  public OtlpHttpSpanExporterBuilder setHeaders(Supplier<Map<String, String>> headerSupplier) {
+    delegate.setHeadersSupplier(headerSupplier);
     return this;
   }