Instrument Netty addTask to ensure complete coverage of async Runnables
Newer versions of Netty introduce variants like execute(Runnable, boolean) which
aren't covered by the core execute(Runnable) instrumentation.  Fortunately they all
flow through to addTask(Runnable), which allows us to carry the context through properly.
MODIFY
instrumentation/java-concurrent/src/main/java/io/opentelemetry/instrumentation/auto/javaconcurrent/JavaExecutorInstrumentation.java
instrumentation/java-concurrent/src/main/java/io/opentelemetry/instrumentation/auto/javaconcurrent/JavaExecutorInstrumentation.java
@@ -50,6 +50,11 @@ public final class JavaExecutorInstrumentation extends AbstractExecutorInstrumen
transformers.put(
named("execute").and(takesArgument(0, Runnable.class)).and(takesArguments(1)),
JavaExecutorInstrumentation.class.getName() + "$SetExecuteRunnableStateAdvice");
+ // Netty uses addTask as the acutal core of their submission; there are non-standard variations
+ // like execute(Runnable,boolean) that aren't caught by standard instrumentation
+ transformers.put(
+ named("addTask").and(takesArgument(0, Runnable.class)).and(takesArguments(1)),
+ JavaExecutorInstrumentation.class.getName() + "$SetExecuteRunnableStateAdvice");
transformers.put(
named("execute").and(takesArgument(0, ForkJoinTask.class)),
JavaExecutorInstrumentation.class.getName() + "$SetJavaForkJoinStateAdvice");

MODIFY
instrumentation/netty/netty-4.1/src/test/groovy/Netty41ClientTest.groovy
instrumentation/netty/netty-4.1/src/test/groovy/Netty41ClientTest.groovy
@@ -8,13 +8,24 @@ import static io.opentelemetry.auto.test.utils.TraceUtils.basicSpan
import static io.opentelemetry.auto.test.utils.TraceUtils.runUnderTrace
import static org.asynchttpclient.Dsl.asyncHttpClient
+import io.netty.bootstrap.Bootstrap
+import io.netty.buffer.Unpooled
import io.netty.channel.AbstractChannel
import io.netty.channel.Channel
import io.netty.channel.ChannelHandler
import io.netty.channel.ChannelHandlerContext
import io.netty.channel.ChannelInitializer
+import io.netty.channel.ChannelPipeline
+import io.netty.channel.EventLoopGroup
import io.netty.channel.embedded.EmbeddedChannel
+import io.netty.channel.nio.NioEventLoopGroup
+import io.netty.channel.socket.SocketChannel
+import io.netty.channel.socket.nio.NioSocketChannel
+import io.netty.handler.codec.http.DefaultFullHttpRequest
import io.netty.handler.codec.http.HttpClientCodec
+import io.netty.handler.codec.http.HttpHeaderNames
+import io.netty.handler.codec.http.HttpMethod
+import io.netty.handler.codec.http.HttpVersion
import io.opentelemetry.auto.test.base.HttpClientTest
import io.opentelemetry.instrumentation.auto.netty.v4_1.client.HttpClientTracingHandler
import java.util.concurrent.ExecutionException
@@ -70,6 +81,68 @@ class Netty41ClientTest extends HttpClientTest {
return false
}
+ def "test connection reuse and second request with lazy execute"() {
+ setup:
+ //Create a simple Netty pipeline
+ EventLoopGroup group = new NioEventLoopGroup()
+ Bootstrap b = new Bootstrap()
+ b.group(group)
+ .channel(NioSocketChannel)
+ .handler(new ChannelInitializer<SocketChannel>() {
+ @Override
+ protected void initChannel(SocketChannel socketChannel) throws Exception {
+ ChannelPipeline pipeline = socketChannel.pipeline()
+ pipeline.addLast(new HttpClientCodec())
+ }
+ })
+ def request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.GET, server.address.resolve("/success").toString(), Unpooled.EMPTY_BUFFER)
+ request.headers().set(HttpHeaderNames.HOST, server.address.host)
+ request.headers().set(HttpHeaderNames.USER_AGENT, userAgent())
+ Channel ch = null
+
+ when:
+ // note that this is a purely asynchronous request
+ runUnderTrace("parent1") {
+ ch = b.connect(server.address.host, server.address.port).sync().channel()
+ ch.write(request)
+ ch.flush()
+ }
+ // This is a cheap/easy way to block/ensure that the first request has finished and check reported spans midway through
+ // the complex sequence of events
+ assertTraces(1) {
+ trace(0, 3) {
+ basicSpan(it, 0, "parent1")
+ clientSpan(it, 1, span(0))
+ serverSpan(it, 2, span(1))
+ }
+ }
+
+ then:
+ // now run a second request through the same channel
+ runUnderTrace("parent2") {
+ ch.write(request)
+ ch.flush()
+ }
+
+ assertTraces(2) {
+ trace(0, 3) {
+ basicSpan(it, 0, "parent1")
+ clientSpan(it, 1, span(0))
+ serverSpan(it, 2, span(1))
+ }
+ trace(1, 3) {
+ basicSpan(it, 0, "parent2")
+ clientSpan(it, 1, span(0))
+ serverSpan(it, 2, span(1))
+ }
+ }
+
+
+ cleanup:
+ group.shutdownGracefully()
+ }
+
+
def "connection error (unopened port)"() {
given:
def uri = new URI("http://localhost:$UNUSABLE_PORT/")

Instrument Netty addTask to ensure complete coverage of async Runnables  (#1348)

Newer versions of Netty introduce variants like execute(Runnable, boolean) which
aren't covered by the core execute(Runnable) instrumentation.  Fortunately they all
flow through to addTask(Runnable), which allows us to carry the context through properly.