Reactor Netty: emit actual HTTP client spans spans on connection errors

instrumentation/netty/netty-4-common/library/src/main/java/io/opentelemetry/instrumentation/netty/v4/common/internal/client/NettyClientInstrumenterFactory.java

@@ -34,22 +34,22 @@ public final class NettyClientInstrumenterFactory {
 
   private final OpenTelemetry openTelemetry;
   private final String instrumentationName;
-  private final boolean connectionTelemetryEnabled;
-  private final boolean sslTelemetryEnabled;
+  private final NettyConnectionInstrumentationFlag connectionTelemetryState;
+  private final NettyConnectionInstrumentationFlag sslTelemetryState;
   private final Map<String, String> peerServiceMapping;
   private final boolean emitExperimentalHttpClientMetrics;
 
   public NettyClientInstrumenterFactory(
       OpenTelemetry openTelemetry,
       String instrumentationName,
-      boolean connectionTelemetryEnabled,
-      boolean sslTelemetryEnabled,
+      NettyConnectionInstrumentationFlag connectionTelemetryState,
+      NettyConnectionInstrumentationFlag sslTelemetryState,
       Map<String, String> peerServiceMapping,
       boolean emitExperimentalHttpClientMetrics) {
     this.openTelemetry = openTelemetry;
     this.instrumentationName = instrumentationName;
-    this.connectionTelemetryEnabled = connectionTelemetryEnabled;
-    this.sslTelemetryEnabled = sslTelemetryEnabled;
+    this.connectionTelemetryState = connectionTelemetryState;
+    this.sslTelemetryState = sslTelemetryState;
     this.peerServiceMapping = peerServiceMapping;
     this.emitExperimentalHttpClientMetrics = emitExperimentalHttpClientMetrics;
   }
@@ -83,49 +83,38 @@ public Instrumenter<HttpRequestAndChannel, HttpResponse> createHttpInstrumenter(
   }
 
   public NettyConnectionInstrumenter createConnectionInstrumenter() {
-    InstrumenterBuilder<NettyConnectionRequest, Channel> instrumenterBuilder =
+    if (connectionTelemetryState == NettyConnectionInstrumentationFlag.DISABLED) {
+      return NoopConnectionInstrumenter.INSTANCE;
+    }
+
+    boolean connectionTelemetryFullyEnabled =
+        connectionTelemetryState == NettyConnectionInstrumentationFlag.ENABLED;
+
+    Instrumenter<NettyConnectionRequest, Channel> instrumenter =
         Instrumenter.<NettyConnectionRequest, Channel>builder(
                 openTelemetry, instrumentationName, NettyConnectionRequest::spanName)
             .addAttributesExtractor(
                 PeerServiceAttributesExtractor.create(
-                    NettyConnectHttpAttributesGetter.INSTANCE, peerServiceMapping));
-
-    if (connectionTelemetryEnabled) {
-      // TODO: this will most likely be no longer true with the new semconv, since the connection
-      // phase happens *before* the actual HTTP request is sent over the wire
-      // TODO (mateusz): refactor this after reactor-netty is fully converted to low-level HTTP
-      // instrumentation
-      // when the connection telemetry is enabled, we do not want these CONNECT spans to be
-      // suppressed by higher-level HTTP spans - this would happen in the reactor-netty
-      // instrumentation
-      // the solution for this is to deliberately avoid using the HTTP extractor and use the plain
-      // net attributes extractor instead
-      instrumenterBuilder.addAttributesExtractor(
-          NetClientAttributesExtractor.create(NettyConnectHttpAttributesGetter.INSTANCE));
-    } else {
-      // when the connection telemetry is not enabled, netty creates CONNECT spans whenever a
-      // connection error occurs - because there is no HTTP span in that scenario, if raw netty
-      // connection occurs before an HTTP message is even formed
-      // we don't want that span when a higher-level HTTP library (like reactor-netty or async http
-      // client) is used, the connection phase is a part of the HTTP span for these
-      // for that to happen, the CONNECT span will "pretend" to be a full HTTP span when connection
-      // telemetry is off
-      instrumenterBuilder.addAttributesExtractor(
-          HttpClientAttributesExtractor.create(NettyConnectHttpAttributesGetter.INSTANCE));
-    }
-
-    Instrumenter<NettyConnectionRequest, Channel> instrumenter =
-        instrumenterBuilder.buildInstrumenter(
-            connectionTelemetryEnabled
-                ? SpanKindExtractor.alwaysInternal()
-                : SpanKindExtractor.alwaysClient());
+                    NettyConnectHttpAttributesGetter.INSTANCE, peerServiceMapping))
+            .addAttributesExtractor(
+                HttpClientAttributesExtractor.create(NettyConnectHttpAttributesGetter.INSTANCE))
+            .buildInstrumenter(
+                connectionTelemetryFullyEnabled
+                    ? SpanKindExtractor.alwaysInternal()
+                    : SpanKindExtractor.alwaysClient());
 
-    return connectionTelemetryEnabled
+    return connectionTelemetryFullyEnabled
         ? new NettyConnectionInstrumenterImpl(instrumenter)
         : new NettyErrorOnlyConnectionInstrumenter(instrumenter);
   }
 
   public NettySslInstrumenter createSslInstrumenter() {
+    if (sslTelemetryState == NettyConnectionInstrumentationFlag.DISABLED) {
+      return NoopSslInstrumenter.INSTANCE;
+    }
+
+    boolean sslTelemetryFullyEnabled =
+        sslTelemetryState == NettyConnectionInstrumentationFlag.ENABLED;
     NettySslNetAttributesGetter netAttributesGetter = new NettySslNetAttributesGetter();
     Instrumenter<NettySslRequest, Void> instrumenter =
         Instrumenter.<NettySslRequest, Void>builder(
@@ -134,11 +123,11 @@ public NettySslInstrumenter createSslInstrumenter() {
             .addAttributesExtractor(
                 PeerServiceAttributesExtractor.create(netAttributesGetter, peerServiceMapping))
             .buildInstrumenter(
-                sslTelemetryEnabled
+                sslTelemetryFullyEnabled
                     ? SpanKindExtractor.alwaysInternal()
                     : SpanKindExtractor.alwaysClient());
 
-    return sslTelemetryEnabled
+    return sslTelemetryFullyEnabled
         ? new NettySslInstrumenterImpl(instrumenter)
         : new NettySslErrorOnlyInstrumenter(instrumenter);
   }

instrumentation/netty/netty-4-common/library/src/main/java/io/opentelemetry/instrumentation/netty/v4/common/internal/client/NettyConnectionInstrumentationFlag.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.netty.v4.common.internal.client;
+
+/**
+ * This class is internal and is hence not for public use. Its APIs are unstable and can change at
+ * any time.
+ */
+public enum NettyConnectionInstrumentationFlag {
+  ENABLED,
+  ERROR_ONLY,
+  DISABLED;
+
+  public static NettyConnectionInstrumentationFlag enabledOrErrorOnly(boolean b) {
+    return b ? ENABLED : ERROR_ONLY;
+  }
+}

instrumentation/netty/netty-4-common/library/src/main/java/io/opentelemetry/instrumentation/netty/v4/common/internal/client/NoopConnectionInstrumenter.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.netty.v4.common.internal.client;
+
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.netty.channel.Channel;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.instrumentation.netty.common.internal.NettyConnectionRequest;
+import javax.annotation.Nullable;
+
+enum NoopConnectionInstrumenter implements NettyConnectionInstrumenter {
+  INSTANCE;
+
+  @Override
+  public boolean shouldStart(Context parentContext, NettyConnectionRequest request) {
+    return false;
+  }
+
+  @CanIgnoreReturnValue
+  @Override
+  public Context start(Context parentContext, NettyConnectionRequest request) {
+    return parentContext;
+  }
+
+  @Override
+  public void end(
+      Context context,
+      NettyConnectionRequest request,
+      Channel channel,
+      @Nullable Throwable error) {}
+}

instrumentation/netty/netty-4-common/library/src/main/java/io/opentelemetry/instrumentation/netty/v4/common/internal/client/NoopSslInstrumenter.java

@@ -0,0 +1,28 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.netty.v4.common.internal.client;
+
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import io.opentelemetry.context.Context;
+import javax.annotation.Nullable;
+
+enum NoopSslInstrumenter implements NettySslInstrumenter {
+  INSTANCE;
+
+  @Override
+  public boolean shouldStart(Context parentContext, NettySslRequest request) {
+    return false;
+  }
+
+  @CanIgnoreReturnValue
+  @Override
+  public Context start(Context parentContext, NettySslRequest request) {
+    return parentContext;
+  }
+
+  @Override
+  public void end(Context context, NettySslRequest request, @Nullable Throwable error) {}
+}

instrumentation/netty/netty-4.0/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/netty/v4_0/client/NettyClientSingletons.java

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.javaagent.instrumentation.netty.v4_0.client;
 
+import static io.opentelemetry.instrumentation.netty.v4.common.internal.client.NettyConnectionInstrumentationFlag.enabledOrErrorOnly;
+
 import io.netty.handler.codec.http.HttpResponse;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
@@ -43,8 +45,8 @@ public final class NettyClientSingletons {
         new NettyClientInstrumenterFactory(
             GlobalOpenTelemetry.get(),
             "io.opentelemetry.netty-4.0",
-            connectionTelemetryEnabled,
-            sslTelemetryEnabled,
+            enabledOrErrorOnly(connectionTelemetryEnabled),
+            enabledOrErrorOnly(sslTelemetryEnabled),
             CommonConfig.get().getPeerServiceMapping(),
             CommonConfig.get().shouldEmitExperimentalHttpClientMetrics());
     INSTRUMENTER =

instrumentation/netty/netty-4.1/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/netty/v4_1/NettyClientSingletons.java

@@ -5,6 +5,8 @@
 
 package io.opentelemetry.javaagent.instrumentation.netty.v4_1;
 
+import static io.opentelemetry.instrumentation.netty.v4.common.internal.client.NettyConnectionInstrumentationFlag.enabledOrErrorOnly;
+
 import io.netty.handler.codec.http.HttpResponse;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.instrumentation.api.instrumenter.Instrumenter;
@@ -43,8 +45,8 @@ public final class NettyClientSingletons {
         new NettyClientInstrumenterFactory(
             GlobalOpenTelemetry.get(),
             "io.opentelemetry.netty-4.1",
-            connectionTelemetryEnabled,
-            sslTelemetryEnabled,
+            enabledOrErrorOnly(connectionTelemetryEnabled),
+            enabledOrErrorOnly(sslTelemetryEnabled),
             CommonConfig.get().getPeerServiceMapping(),
             CommonConfig.get().shouldEmitExperimentalHttpClientMetrics());
     INSTRUMENTER =

instrumentation/netty/netty-4.1/library/src/main/java/io/opentelemetry/instrumentation/netty/v4_1/NettyClientTelemetryBuilder.java

@@ -12,6 +12,7 @@
 import io.opentelemetry.instrumentation.api.instrumenter.http.HttpClientAttributesExtractorBuilder;
 import io.opentelemetry.instrumentation.netty.v4.common.HttpRequestAndChannel;
 import io.opentelemetry.instrumentation.netty.v4.common.internal.client.NettyClientInstrumenterFactory;
+import io.opentelemetry.instrumentation.netty.v4.common.internal.client.NettyConnectionInstrumentationFlag;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -111,8 +112,8 @@ public NettyClientTelemetry build() {
         new NettyClientInstrumenterFactory(
                 openTelemetry,
                 "io.opentelemetry.netty-4.1",
-                false,
-                false,
+                NettyConnectionInstrumentationFlag.DISABLED,
+                NettyConnectionInstrumentationFlag.DISABLED,
                 Collections.emptyMap(),
                 emitExperimentalHttpClientMetrics)
             .createHttpInstrumenter(extractorConfigurer, additionalAttributesExtractors));

instrumentation/reactor/reactor-netty/reactor-netty-1.0/javaagent-unit-tests/build.gradle.kts

@@ -4,4 +4,5 @@ plugins {
 
 dependencies {
   testImplementation(project(":instrumentation:reactor:reactor-netty:reactor-netty-1.0:javaagent"))
+  testImplementation("io.projectreactor.netty:reactor-netty-http:1.0.0")
 }

instrumentation/reactor/reactor-netty/reactor-netty-1.0/javaagent-unit-tests/src/test/java/io/opentelemetry/javaagent/instrumentation/reactornetty/v1_0/FailedRequestWithUrlMakerTest.java

@@ -0,0 +1,92 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.javaagent.instrumentation.reactornetty.v1_0;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.when;
+
+import java.net.InetSocketAddress;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.junit.jupiter.api.extension.ExtensionContext;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import reactor.netty.http.client.HttpClientConfig;
+import reactor.netty.http.client.HttpClientRequest;
+
+@ExtendWith(MockitoExtension.class)
+class FailedRequestWithUrlMakerTest {
+
+  @Mock HttpClientConfig config;
+  @Mock HttpClientRequest originalRequest;
+
+  @Test
+  void shouldUseAbsoluteUri() {
+    when(config.uri()).thenReturn("https://opentelemetry.io");
+
+    HttpClientRequest request = FailedRequestWithUrlMaker.create(config, originalRequest);
+
+    assertThat(request.resourceUrl()).isEqualTo("https://opentelemetry.io");
+  }
+
+  @ParameterizedTest
+  @ValueSource(strings = {"https://opentelemetry.io", "https://opentelemetry.io/"})
+  void shouldPrependBaseUrl(String baseUrl) {
+    when(config.baseUrl()).thenReturn(baseUrl);
+    when(config.uri()).thenReturn("/docs");
+
+    HttpClientRequest request = FailedRequestWithUrlMaker.create(config, originalRequest);
+
+    assertThat(request.resourceUrl()).isEqualTo("https://opentelemetry.io/docs");
+  }
+
+  @Test
+  void shouldPrependRemoteAddress() {
+    when(config.baseUrl()).thenReturn("/");
+    when(config.uri()).thenReturn("/docs");
+    Supplier<InetSocketAddress> remoteAddress =
+        () -> InetSocketAddress.createUnresolved("opentelemetry.io", 8080);
+    doReturn(remoteAddress).when(config).remoteAddress();
+    when(config.isSecure()).thenReturn(true);
+
+    HttpClientRequest request = FailedRequestWithUrlMaker.create(config, originalRequest);
+
+    assertThat(request.resourceUrl()).isEqualTo("https://opentelemetry.io:8080/docs");
+  }
+
+  @ParameterizedTest
+  @ArgumentsSource(DefaultPortsArguments.class)
+  void shouldSkipDefaultPorts(int port, boolean isSecure) {
+    when(config.baseUrl()).thenReturn("/");
+    when(config.uri()).thenReturn("/docs");
+    Supplier<InetSocketAddress> remoteAddress =
+        () -> InetSocketAddress.createUnresolved("opentelemetry.io", port);
+    doReturn(remoteAddress).when(config).remoteAddress();
+    when(config.isSecure()).thenReturn(isSecure);
+
+    HttpClientRequest request = FailedRequestWithUrlMaker.create(config, originalRequest);
+
+    assertThat(request.resourceUrl())
+        .isEqualTo((isSecure ? "https" : "http") + "://opentelemetry.io/docs");
+  }
+
+  static final class DefaultPortsArguments implements ArgumentsProvider {
+
+    @Override
+    public Stream<? extends Arguments> provideArguments(ExtensionContext extensionContext) {
+      return Stream.of(arguments(80, false), arguments(443, true));
+    }
+  }
+}