Add OTLP header supplier configuration option

docs/apidiffs/current_vs_latest/opentelemetry-exporter-otlp.txt

@@ -1,2 +1,19 @@
 Comparing source compatibility of  against 
-No changes.
+***  MODIFIED CLASS: PUBLIC FINAL io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++  NEW METHOD: PUBLIC(+) io.opentelemetry.exporter.otlp.http.logs.OtlpHttpLogRecordExporterBuilder setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.util.List<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 setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.util.List<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 setHeaders(java.util.function.Supplier<java.util.Map<java.lang.String,java.util.List<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.util.List<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.util.List<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.util.List<java.lang.String>>>)

exporters/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 @@
   private long timeoutNanos;
   private URI endpoint;
   private boolean compressionEnabled = false;
-  private final Map<String, String> headers = new HashMap<>();
+  private final Map<String, List<String>> constantHeaders = new HashMap<>();
+  private Supplier<Map<String, List<String>>> headerSupplier = Collections::emptyMap;
   private TlsConfigHelper tlsConfigHelper = new TlsConfigHelper();
   @Nullable private RetryPolicy retryPolicy;
   private Supplier<MeterProvider> meterProviderSupplier = GlobalOpenTelemetry::getMeterProvider;
@@ -113,8 +117,14 @@
     return this;
   }
 
-  public GrpcExporterBuilder<T> addHeader(String key, String value) {
-    headers.put(key, value);
+  public GrpcExporterBuilder<T> addConstantHeader(String key, String value) {
+    constantHeaders.computeIfAbsent(key, unused -> new ArrayList<>()).add(value);
+    return this;
+  }
+
+  public GrpcExporterBuilder<T> setHeadersSupplier(
+      Supplier<Map<String, List<String>>> headerSupplier) {
+    this.headerSupplier = headerSupplier;
     return this;
   }
 
@@ -142,7 +152,8 @@
     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 +164,34 @@
   }
 
   public GrpcExporter<T> build() {
+    Supplier<Map<String, List<String>>> headerSupplier =
+        () -> {
+          Map<String, List<String>> result = new HashMap<>();
+          Map<String, List<String>> supplierResult = this.headerSupplier.get();
+          if (supplierResult != null) {
+            result.putAll(supplierResult);
+          }
+          constantHeaders.forEach(
+              (key, value) ->
+                  result.merge(
+                      key,
+                      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 @@
     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, List<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);

exporters/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,

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

@@ -15,8 +15,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;
@@ -48,7 +50,8 @@
   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 Map<String, List<String>> constantHeaders = new HashMap<>();
+  private Supplier<Map<String, List<String>>> headerSupplier = Collections::emptyMap;
 
   private TlsConfigHelper tlsConfigHelper = new TlsConfigHelper();
   @Nullable private RetryPolicy retryPolicy;
@@ -82,11 +85,14 @@
     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.computeIfAbsent(key, unused -> new ArrayList<>()).add(value);
+    return this;
+  }
+
+  public HttpExporterBuilder<T> setHeadersSupplier(
+      Supplier<Map<String, List<String>>> headerSupplier) {
+    this.headerSupplier = headerSupplier;
     return this;
   }
 
@@ -134,9 +140,8 @@
     copy.timeoutNanos = timeoutNanos;
     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();
@@ -147,8 +152,25 @@
   }
 
   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, List<String>> supplierResult = this.headerSupplier.get();
+          if (supplierResult != null) {
+            result.putAll(supplierResult);
+          }
+          constantHeaders.forEach(
+              (key, value) ->
+                  result.merge(
+                      key,
+                      value,
+                      (v1, v2) -> {
+                        List<String> merged = new ArrayList<>(v1);
+                        merged.addAll(v2);
+                        return merged;
+                      }));
+          return result;
+        };
 
     HttpSenderProvider httpSenderProvider = resolveHttpSenderProvider();
     HttpSender httpSender =
@@ -178,11 +200,13 @@
     joiner.add("timeoutNanos=" + timeoutNanos);
     joiner.add("compressionEnabled=" + compressionEnabled);
     joiner.add("exportAsJson=" + exportAsJson);
+    StringJoiner headersJoiner = new StringJoiner(", ", "Headers{", "}");
+    constantHeaders.forEach((key, value) -> headersJoiner.add(key + "=OBFUSCATED"));
+    Map<String, List<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);
     }

exporters/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;
@@ -28,7 +29,7 @@ HttpSender createSender(
       boolean compressionEnabled,
       String contentType,
       long timeoutNanos,
-      Supplier<Map<String, String>> headerSupplier,
+      Supplier<Map<String, List<String>>> headerSupplier,
       @Nullable Authenticator authenticator,
       @Nullable RetryPolicy retryPolicy,
       @Nullable SSLContext sslContext,

exporters/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),

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

@@ -15,6 +15,8 @@
 import io.opentelemetry.exporter.otlp.internal.OtlpUserAgent;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.time.Duration;
+import java.util.List;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import javax.net.ssl.SSLContext;
@@ -33,7 +35,7 @@ public final class OtlpHttpLogRecordExporterBuilder {
 
   OtlpHttpLogRecordExporterBuilder(HttpExporterBuilder<LogsRequestMarshaler> delegate) {
     this.delegate = delegate;
-    OtlpUserAgent.addUserAgentHeader(delegate::addHeader);
+    OtlpUserAgent.addUserAgentHeader(delegate::addConstantHeaders);
   }
 
   OtlpHttpLogRecordExporterBuilder() {
@@ -83,9 +85,22 @@ 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, List<String>>> headerSupplier) {
+    delegate.setHeadersSupplier(headerSupplier);
     return this;
   }
 

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

@@ -18,7 +18,10 @@
 import io.opentelemetry.sdk.metrics.export.DefaultAggregationSelector;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.time.Duration;
+import java.util.List;
+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 +47,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() {
@@ -94,9 +97,22 @@ 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, List<String>>> headerSupplier) {
+    delegate.setHeadersSupplier(headerSupplier);
     return this;
   }