Allow overriding OTLP gRPC authority. (#4514)

docs/apidiffs/current_vs_latest/opentelemetry-sdk-extension-autoconfigure-spi.txt

@@ -1,4 +1,14 @@
 Comparing source compatibility of  against 
+***! MODIFIED INTERFACE: PUBLIC ABSTRACT io.opentelemetry.sdk.autoconfigure.spi.ConfigProperties  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++! NEW METHOD: PUBLIC(+) boolean getBoolean(java.lang.String, boolean)
+	+++! NEW METHOD: PUBLIC(+) double getDouble(java.lang.String, double)
+	+++! NEW METHOD: PUBLIC(+) java.time.Duration getDuration(java.lang.String, java.time.Duration)
+	+++! NEW METHOD: PUBLIC(+) int getInt(java.lang.String, int)
+	+++! NEW METHOD: PUBLIC(+) java.util.List getList(java.lang.String, java.util.List)
+	+++! NEW METHOD: PUBLIC(+) long getLong(java.lang.String, long)
+	+++! NEW METHOD: PUBLIC(+) java.util.Map getMap(java.lang.String, java.util.Map)
+	+++! NEW METHOD: PUBLIC(+) java.lang.String getString(java.lang.String, java.lang.String)
 +++  NEW INTERFACE: PUBLIC(+) ABSTRACT(+) io.opentelemetry.sdk.autoconfigure.spi.metrics.ConfigurableMetricExporterProvider  (not serializable)
 	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
 	+++  NEW SUPERCLASS: java.lang.Object

exporters/jaeger/src/main/java/io/opentelemetry/exporter/jaeger/MarshalerCollectorServiceGrpc.java

@@ -15,6 +15,7 @@
 import io.opentelemetry.exporter.internal.grpc.MarshalerInputStream;
 import io.opentelemetry.exporter.internal.grpc.MarshalerServiceStub;
 import java.io.InputStream;
+import javax.annotation.Nullable;
 
 // Adapted from the protoc generated code for CollectorServiceGrpc.
 final class MarshalerCollectorServiceGrpc {
@@ -56,8 +57,11 @@ public PostSpansResponse parse(InputStream stream) {
               .setResponseMarshaller(RESPONSE_MARSHALER)
               .build();
 
-  static CollectorServiceFutureStub newFutureStub(Channel channel) {
-    return CollectorServiceFutureStub.newStub(CollectorServiceFutureStub::new, channel);
+  static CollectorServiceFutureStub newFutureStub(
+      Channel channel, @Nullable String authorityOverride) {
+    return CollectorServiceFutureStub.newStub(
+        (c, options) -> new CollectorServiceFutureStub(c, options.withAuthority(authorityOverride)),
+        channel);
   }
 
   static final class CollectorServiceFutureStub

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

@@ -16,6 +16,7 @@
 import io.opentelemetry.exporter.internal.grpc.MarshalerServiceStub;
 import io.opentelemetry.exporter.internal.otlp.metrics.MetricsRequestMarshaler;
 import java.io.InputStream;
+import javax.annotation.Nullable;
 
 // Adapted from the protoc generated code for MetricsServiceGrpc.
 final class MarshalerMetricsServiceGrpc {
@@ -59,8 +60,11 @@ public ExportMetricsServiceResponse parse(InputStream stream) {
               .setResponseMarshaller(RESPONSE_MARSHALER)
               .build();
 
-  static MetricsServiceFutureStub newFutureStub(Channel channel) {
-    return MetricsServiceFutureStub.newStub(MetricsServiceFutureStub::new, channel);
+  static MetricsServiceFutureStub newFutureStub(
+      Channel channel, @Nullable String authorityOverride) {
+    return MetricsServiceFutureStub.newStub(
+        (c, options) -> new MetricsServiceFutureStub(c, options.withAuthority(authorityOverride)),
+        channel);
   }
 
   static final class MetricsServiceFutureStub

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

@@ -12,6 +12,7 @@
 import io.opentelemetry.exporter.internal.grpc.MarshalerServiceStub;
 import io.opentelemetry.exporter.internal.otlp.traces.TraceRequestMarshaler;
 import java.io.InputStream;
+import javax.annotation.Nullable;
 
 // Adapted from the protoc generated code for TraceServiceGrpc.
 final class MarshalerTraceServiceGrpc {
@@ -53,8 +54,11 @@ public ExportTraceServiceResponse parse(InputStream stream) {
               .setResponseMarshaller(RESPONSE_MARSHALER)
               .build();
 
-  static TraceServiceFutureStub newFutureStub(io.grpc.Channel channel) {
-    return TraceServiceFutureStub.newStub(TraceServiceFutureStub::new, channel);
+  static TraceServiceFutureStub newFutureStub(
+      io.grpc.Channel channel, @Nullable String authorityOverride) {
+    return TraceServiceFutureStub.newStub(
+        (c, options) -> new TraceServiceFutureStub(c, options.withAuthority(authorityOverride)),
+        channel);
   }
 
   static final class TraceServiceFutureStub

exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/grpc/DefaultGrpcExporterBuilder.java

@@ -20,7 +20,7 @@
 import java.net.URI;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java.util.function.BiFunction;
 import javax.annotation.Nullable;
 import javax.net.ssl.SSLException;
 
@@ -35,7 +35,7 @@ public final class DefaultGrpcExporterBuilder<T extends Marshaler>
 
   private final String exporterName;
   private final String type;
-  private final Function<ManagedChannel, MarshalerServiceStub<T, ?, ?>> stubFactory;
+  private final BiFunction<ManagedChannel, String, MarshalerServiceStub<T, ?, ?>> stubFactory;
   private final String grpcServiceName;
 
   @Nullable private ManagedChannel channel;
@@ -47,14 +47,15 @@ public final class DefaultGrpcExporterBuilder<T extends Marshaler>
   @Nullable private byte[] privateKeyPem;
   @Nullable private byte[] certificatePem;
   @Nullable RetryPolicy retryPolicy;
+  @Nullable String authorityOverride;
   private MeterProvider meterProvider = MeterProvider.noop();
 
   /** Creates a new {@link DefaultGrpcExporterBuilder}. */
   // Visible for testing
   public DefaultGrpcExporterBuilder(
       String exporterName,
       String type,
-      Function<ManagedChannel, MarshalerServiceStub<T, ?, ?>> stubFactory,
+      BiFunction<ManagedChannel, String, MarshalerServiceStub<T, ?, ?>> stubFactory,
       long defaultTimeoutSecs,
       URI defaultEndpoint,
       String grpcServiceName) {
@@ -110,6 +111,10 @@ public GrpcExporterBuilder<T> setClientTls(byte[] privateKeyPem, byte[] certific
 
   @Override
   public DefaultGrpcExporterBuilder<T> addHeader(String key, String value) {
+    if (key.equals("host")) {
+      authorityOverride = value;
+      return this;
+    }
     if (metadata == null) {
       metadata = new Metadata();
     }
@@ -167,7 +172,7 @@ public GrpcExporter<T> build() {
 
     Codec codec = compressionEnabled ? new Codec.Gzip() : Codec.Identity.NONE;
     MarshalerServiceStub<T, ?, ?> stub =
-        stubFactory.apply(channel).withCompression(codec.getMessageEncoding());
+        stubFactory.apply(channel, authorityOverride).withCompression(codec.getMessageEncoding());
     return new DefaultGrpcExporter<>(
         exporterName, type, channel, stub, meterProvider, timeoutNanos);
   }

exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporter.java

@@ -9,7 +9,7 @@
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import java.net.URI;
-import java.util.function.Function;
+import java.util.function.BiFunction;
 import java.util.function.Supplier;
 
 /**
@@ -26,7 +26,7 @@ static <T extends Marshaler> GrpcExporterBuilder<T> builder(
       String type,
       long defaultTimeoutSecs,
       URI defaultEndpoint,
-      Supplier<Function<ManagedChannel, MarshalerServiceStub<T, ?, ?>>> stubFactory,
+      Supplier<BiFunction<ManagedChannel, String, MarshalerServiceStub<T, ?, ?>>> stubFactory,
       String grpcServiceName,
       String grpcEndpointPath) {
     return GrpcExporterUtil.exporterBuilder(

exporters/otlp/common/src/main/java/io/opentelemetry/exporter/internal/grpc/GrpcExporterUtil.java

@@ -8,7 +8,7 @@
 import io.grpc.ManagedChannel;
 import io.opentelemetry.exporter.internal.marshal.Marshaler;
 import java.net.URI;
-import java.util.function.Function;
+import java.util.function.BiFunction;
 import java.util.function.Supplier;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -35,7 +35,7 @@ static <T extends Marshaler> GrpcExporterBuilder<T> exporterBuilder(
       String type,
       long defaultTimeoutSecs,
       URI defaultEndpoint,
-      Supplier<Function<ManagedChannel, MarshalerServiceStub<T, ?, ?>>> stubFactory,
+      Supplier<BiFunction<ManagedChannel, String, MarshalerServiceStub<T, ?, ?>>> stubFactory,
       String grpcServiceName,
       String grpcEndpointPath) {
     if (USE_OKHTTP) {

exporters/otlp/common/src/test/java/io/opentelemetry/exporter/internal/retry/RetryUtilTest.java

@@ -24,7 +24,7 @@ void setRetryPolicyOnDelegate_DefaultGrpcExporterBuilder() throws URISyntaxExcep
     RetryPolicy retryPolicy = RetryPolicy.getDefault();
     DefaultGrpcExporterBuilder<?> builder =
         new DefaultGrpcExporterBuilder<>(
-            "otlp", "test", unused -> null, 0, new URI("http://localhost"), "test");
+            "otlp", "test", (u1, u2) -> null, 0, new URI("http://localhost"), "test");
 
     RetryUtil.setRetryPolicyOnDelegate(new WithDelegate(builder), retryPolicy);
 

exporters/otlp/logs/src/main/java/io/opentelemetry/exporter/otlp/logs/MarshalerLogsServiceGrpc.java

@@ -16,6 +16,7 @@
 import io.opentelemetry.exporter.internal.grpc.MarshalerServiceStub;
 import io.opentelemetry.exporter.internal.otlp.logs.LogsRequestMarshaler;
 import java.io.InputStream;
+import javax.annotation.Nullable;
 
 // Adapted from the protoc generated code for LogsServiceGrpc.
 final class MarshalerLogsServiceGrpc {
@@ -57,8 +58,10 @@ public ExportLogsServiceResponse parse(InputStream stream) {
               .setResponseMarshaller(RESPONSE_MARSHALER)
               .build();
 
-  static LogsServiceFutureStub newFutureStub(Channel channel) {
-    return LogsServiceFutureStub.newStub(LogsServiceFutureStub::new, channel);
+  static LogsServiceFutureStub newFutureStub(Channel channel, @Nullable String authorityOverride) {
+    return LogsServiceFutureStub.newStub(
+        (c, options) -> new LogsServiceFutureStub(c, options.withAuthority(authorityOverride)),
+        channel);
   }
 
   static final class LogsServiceFutureStub

exporters/otlp/testing-internal/src/main/java/io/opentelemetry/exporter/otlp/testing/internal/AbstractGrpcTelemetryExporterTest.java

@@ -11,6 +11,7 @@
 
 import com.google.protobuf.InvalidProtocolBufferException;
 import com.google.protobuf.Message;
+import com.linecorp.armeria.common.HttpRequest;
 import com.linecorp.armeria.common.grpc.protocol.ArmeriaStatusException;
 import com.linecorp.armeria.server.ServerBuilder;
 import com.linecorp.armeria.server.ServiceRequestContext;
@@ -73,6 +74,9 @@ public abstract class AbstractGrpcTelemetryExporterTest<T, U extends Message> {
 
   private static final AtomicInteger attempts = new AtomicInteger();
 
+  private static final ConcurrentLinkedQueue<HttpRequest> httpRequests =
+      new ConcurrentLinkedQueue<>();
+
   @RegisterExtension
   @Order(1)
   static final SelfSignedCertificateExtension certificate = new SelfSignedCertificateExtension();
@@ -131,6 +135,7 @@ private CollectorService(
 
     @Override
     protected CompletionStage<byte[]> handleMessage(ServiceRequestContext ctx, byte[] message) {
+      httpRequests.add(ctx.request());
       attempts.incrementAndGet();
       T request;
       try {
@@ -177,6 +182,7 @@ void reset() {
     exportedResourceTelemetry.clear();
     grpcErrors.clear();
     attempts.set(0);
+    httpRequests.clear();
   }
 
   @Test
@@ -500,6 +506,27 @@ void nonRetryableError(int code) {
     assertThat(attempts).hasValue(1);
   }
 
+  @Test
+  void overrideHost() {
+    List<T> telemetry = Collections.singletonList(generateFakeTelemetry());
+    TelemetryExporter<T> exporter =
+        exporterBuilder()
+            .setEndpoint(server.httpUri().toString())
+            .addHeader("host", "opentelemetry")
+            .build();
+    try {
+      assertThat(exporter.export(telemetry).join(10, TimeUnit.SECONDS).isSuccess()).isTrue();
+    } finally {
+      exporter.shutdown();
+    }
+    List<U> expectedResourceTelemetry = toProto(telemetry);
+    assertThat(exportedResourceTelemetry).containsExactlyElementsOf(expectedResourceTelemetry);
+
+    assertThat(httpRequests)
+        .singleElement()
+        .satisfies(req -> assertThat(req.authority()).isEqualTo("opentelemetry"));
+  }
+
   @Test
   @SuppressWarnings("PreferJavaTimeOverload")
   void validConfig() {