Test http pipelining (#8403)

instrumentation/akka/akka-http-10.0/javaagent/src/test/scala/io/opentelemetry/javaagent/instrumentation/akkahttp/AkkaHttpServerInstrumentationTestAsync.scala

@@ -6,7 +6,10 @@
 package io.opentelemetry.javaagent.instrumentation.akkahttp
 
 import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension
-import io.opentelemetry.instrumentation.testing.junit.http.HttpServerInstrumentationExtension
+import io.opentelemetry.instrumentation.testing.junit.http.{
+  HttpServerInstrumentationExtension,
+  HttpServerTestOptions
+}
 import org.junit.jupiter.api.extension.RegisterExtension
 
 class AkkaHttpServerInstrumentationTestAsync
@@ -22,4 +25,11 @@ class AkkaHttpServerInstrumentationTestAsync
 
   override protected def stopServer(server: Object): Unit =
     AkkaHttpTestAsyncWebServer.stop()
+
+  override protected def configure(
+      options: HttpServerTestOptions
+  ): Unit = {
+    super.configure(options)
+    options.setTestHttpPipelining(false)
+  }
 }

instrumentation/armeria-1.3/javaagent/src/test/java/io/opentelemetry/javaagent/instrumentation/armeria/v1_3/ArmeriaHttpServerTest.java

@@ -28,5 +28,6 @@ class ArmeriaHttpServerTest extends AbstractArmeriaHttpServerTest {
     super.configure(options);
     options.setHasResponseCustomizer(
         endpoint -> ServerEndpoint.NOT_FOUND != endpoint && ServerEndpoint.EXCEPTION != endpoint);
+    options.setTestHttpPipelining(false);
   }
 }

instrumentation/finatra-2.9/javaagent/src/latestDepTest/scala/io/opentelemetry/javaagent/instrumentation/finatra/FinatraServerLatestTest.scala

@@ -55,6 +55,7 @@ class FinatraServerLatestTest extends AbstractHttpServerTest[HttpServer] {
       override def test(endpoint: ServerEndpoint): Boolean =
         endpoint != ServerEndpoint.NOT_FOUND
     })
+    options.setTestHttpPipelining(false)
   }
 
   override protected def assertHandlerSpan(

instrumentation/finatra-2.9/javaagent/src/test/scala/io/opentelemetry/javaagent/instrumentation/finatra/FinatraServerTest.scala

@@ -56,6 +56,7 @@ class FinatraServerTest extends AbstractHttpServerTest[HttpServer] {
       override def test(endpoint: ServerEndpoint): Boolean =
         endpoint != ServerEndpoint.NOT_FOUND
     })
+    options.setTestHttpPipelining(false)
   }
 
   override protected def assertHandlerSpan(

instrumentation/grizzly-2.0/javaagent/src/test/groovy/GrizzlyAsyncTest.groovy

@@ -39,6 +39,11 @@ class GrizzlyAsyncTest extends GrizzlyTest {
     false
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   boolean verifyServerSpanEndTime() {
     // server spans are ended inside of the JAX-RS controller spans

instrumentation/grizzly-2.0/javaagent/src/test/groovy/GrizzlyFilterchainServerTest.groovy

@@ -84,6 +84,11 @@ class GrizzlyFilterchainServerTest extends HttpServerTest<HttpServer> implements
     false
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   boolean verifyServerSpanEndTime() {
     // server spans are ended inside of the controller spans

instrumentation/grizzly-2.0/javaagent/src/test/groovy/GrizzlyTest.groovy

@@ -68,6 +68,11 @@ class GrizzlyTest extends HttpServerTest<HttpServer> implements AgentTestTrait {
     false
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   static class SimpleExceptionMapper implements ExceptionMapper<Throwable> {
 
     @Override

instrumentation/jaxrs/jaxrs-2.0/jaxrs-2.0-resteasy-3.0/javaagent/src/test/groovy/ResteasyHttpServerTest.groovy

@@ -29,7 +29,13 @@ class ResteasyHttpServerTest extends JaxRsHttpServerTest<UndertowJaxrsServer> {
   }
 
   // resteasy 3.0.x does not support JAX-RS 2.1
+  @Override
   boolean shouldTestCompletableStageAsync() {
     false
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

instrumentation/jaxrs/jaxrs-2.0/jaxrs-2.0-resteasy-3.1/javaagent/src/test/groovy/ResteasyHttpServerTest.groovy

@@ -27,4 +27,9 @@ class ResteasyHttpServerTest extends JaxRsHttpServerTest<UndertowJaxrsServer> {
   void stopServer(UndertowJaxrsServer server) {
     server.stop()
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

instrumentation/netty/netty-4.1/testing/src/main/java/io/opentelemetry/instrumentation/netty/v4_1/AbstractNetty41ServerTest.java

@@ -62,6 +62,7 @@ public abstract class AbstractNetty41ServerTest extends AbstractHttpServerTest<E
         unused ->
             Sets.difference(
                 DEFAULT_HTTP_ATTRIBUTES, Collections.singleton(SemanticAttributes.HTTP_ROUTE)));
+    options.setTestHttpPipelining(false);
   }
 
   @Override

instrumentation/play/play-mvc/play-mvc-2.4/javaagent/src/play24Test/groovy/server/PlayServerTest.groovy

@@ -90,6 +90,11 @@ class PlayServerTest extends HttpServerTest<Server> implements AgentTestTrait {
     true
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   boolean verifyServerSpanEndTime() {
     // server spans are ended inside of the controller spans

instrumentation/play/play-mvc/play-mvc-2.4/javaagent/src/test/groovy/server/PlayServerTest.groovy

@@ -85,6 +85,11 @@ class PlayServerTest extends HttpServerTest<Server> implements AgentTestTrait {
     true
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   boolean verifyServerSpanEndTime() {
     // server spans are ended inside of the controller spans

instrumentation/play/play-mvc/play-mvc-2.6/javaagent/src/latestDepTest/groovy/server/PlayServerTest.groovy

@@ -87,6 +87,11 @@ class PlayServerTest extends HttpServerTest<Server> implements AgentTestTrait {
     true
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   void handlerSpan(TraceAssert trace, int index, Object parent, String method = "GET", ServerEndpoint endpoint = SUCCESS) {
     trace.span(index) {

instrumentation/play/play-mvc/play-mvc-2.6/javaagent/src/test/groovy/server/PlayAsyncServerTest.groovy

@@ -98,4 +98,9 @@ class PlayAsyncServerTest extends PlayServerTest {
         .build()
     }
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

instrumentation/ratpack/ratpack-1.4/javaagent/src/test/groovy/server/RatpackForkedHttpServerTest.groovy

@@ -19,4 +19,9 @@ class RatpackForkedHttpServerTest extends AbstractRatpackForkedHttpServerTest im
   boolean hasResponseCustomizer(ServerEndpoint endpoint) {
     true
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

instrumentation/restlet/restlet-1.1/testing/src/main/groovy/io/opentelemetry/instrumentation/restlet/v1_1/AbstractRestletServerTest.groovy

@@ -162,6 +162,11 @@ abstract class AbstractRestletServerTest extends HttpServerTest<Server> {
     false
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   String expectedHttpRoute(ServerEndpoint endpoint) {
     switch (endpoint) {

instrumentation/restlet/restlet-2.0/testing/src/main/groovy/io/opentelemetry/instrumentation/restlet/v2_0/AbstractRestletServerTest.groovy

@@ -175,6 +175,11 @@ abstract class AbstractRestletServerTest extends HttpServerTest<Server> {
     false
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   String expectedHttpRoute(ServerEndpoint endpoint) {
     switch (endpoint) {

instrumentation/spring/spring-webflux/spring-webflux-5.0/javaagent/src/test/java/server/base/SpringWebFluxServerTest.java

@@ -68,5 +68,6 @@ public abstract class SpringWebFluxServerTest
     options.setTestPathParam(true);
     options.setExpectedException(new IllegalStateException(EXCEPTION.getBody()));
     options.setHasHandlerSpan(unused -> true);
+    options.setTestHttpPipelining(false);
   }
 }

instrumentation/vertx/vertx-rx-java-3.5/javaagent/src/latestDepTest/groovy/server/VertxRxCircuitBreakerHttpServerTest.groovy

@@ -162,4 +162,9 @@ class VertxRxCircuitBreakerHttpServerTest extends VertxRxHttpServerTest {
   boolean hasExceptionOnServerSpan(ServerEndpoint endpoint) {
     return endpoint != EXCEPTION && super.hasExceptionOnServerSpan(endpoint)
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

instrumentation/vertx/vertx-rx-java-3.5/javaagent/src/latestDepTest/groovy/server/VertxRxHttpServerTest.groovy

@@ -68,6 +68,11 @@ class VertxRxHttpServerTest extends HttpServerTest<Vertx> implements AgentTestTr
     return false
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   protected Class<AbstractVerticle> verticle() {
     return VertxReactiveWebServer
   }

instrumentation/vertx/vertx-rx-java-3.5/javaagent/src/version35Test/groovy/server/VertxRxCircuitBreakerHttpServerTest.groovy

@@ -161,4 +161,9 @@ class VertxRxCircuitBreakerHttpServerTest extends VertxRxHttpServerTest {
   boolean hasExceptionOnServerSpan(ServerEndpoint endpoint) {
     return endpoint != EXCEPTION && super.hasExceptionOnServerSpan(endpoint)
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

instrumentation/vertx/vertx-rx-java-3.5/javaagent/src/version35Test/groovy/server/VertxRxHttpServerTest.groovy

@@ -62,6 +62,11 @@ class VertxRxHttpServerTest extends HttpServerTest<Vertx> implements AgentTestTr
     return true
   }
 
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
+
   @Override
   boolean verifyServerSpanEndTime() {
     // server spans are ended inside of the controller spans

instrumentation/vertx/vertx-web-3.0/javaagent/src/latestDepTest/groovy/server/VertxLatestHttpServerTest.groovy

@@ -13,4 +13,9 @@ class VertxLatestHttpServerTest extends AbstractVertxHttpServerTest {
   protected Class<? extends AbstractVerticle> verticle() {
     return VertxLatestWebServer
   }
+
+  @Override
+  boolean testHttpPipelining() {
+    false
+  }
 }

testing-common/src/main/groovy/io/opentelemetry/instrumentation/test/base/HttpServerTest.groovy

@@ -144,6 +144,10 @@ abstract class HttpServerTest<SERVER> extends InstrumentationSpecification imple
     true
   }
 
+  boolean testHttpPipelining() {
+    true
+  }
+
   boolean verifyServerSpanEndTime() {
     return true
   }
@@ -217,6 +221,7 @@ abstract class HttpServerTest<SERVER> extends InstrumentationSpecification imple
       options.testPathParam = testPathParam()
       options.testCaptureHttpHeaders = testCapturedHttpHeaders()
       options.testCaptureRequestParameters = testCapturedRequestParameters()
+      options.testHttpPipelining = testHttpPipelining()
     }
 
     // Override trace assertion method. We can call java assertions from groovy but not the other
@@ -314,6 +319,12 @@ abstract class HttpServerTest<SERVER> extends InstrumentationSpecification imple
     junitTest.highConcurrency()
   }
 
+  def "http pipelining test"() {
+    assumeTrue(testHttpPipelining())
+    expect:
+    junitTest.httpPipelining()
+  }
+
   void assertHighConcurrency(int count) {
     def endpoint = INDEXED_CHILD
     assertTraces(count) {

testing-common/src/main/java/io/opentelemetry/instrumentation/testing/junit/http/AbstractHttpServerTest.java

@@ -36,17 +36,36 @@ import io.opentelemetry.semconv.trace.attributes.SemanticAttributes;
 import io.opentelemetry.testing.internal.armeria.common.AggregatedHttpRequest;
 import io.opentelemetry.testing.internal.armeria.common.AggregatedHttpResponse;
 import io.opentelemetry.testing.internal.armeria.common.HttpData;
+import io.opentelemetry.testing.internal.armeria.common.HttpHeaderNames;
 import io.opentelemetry.testing.internal.armeria.common.HttpMethod;
 import io.opentelemetry.testing.internal.armeria.common.HttpRequest;
 import io.opentelemetry.testing.internal.armeria.common.HttpRequestBuilder;
 import io.opentelemetry.testing.internal.armeria.common.MediaType;
 import io.opentelemetry.testing.internal.armeria.common.QueryParams;
 import io.opentelemetry.testing.internal.armeria.common.RequestHeaders;
+import io.opentelemetry.testing.internal.io.netty.bootstrap.Bootstrap;
+import io.opentelemetry.testing.internal.io.netty.buffer.Unpooled;
+import io.opentelemetry.testing.internal.io.netty.channel.Channel;
+import io.opentelemetry.testing.internal.io.netty.channel.ChannelHandlerContext;
+import io.opentelemetry.testing.internal.io.netty.channel.ChannelInitializer;
+import io.opentelemetry.testing.internal.io.netty.channel.ChannelOption;
+import io.opentelemetry.testing.internal.io.netty.channel.ChannelPipeline;
+import io.opentelemetry.testing.internal.io.netty.channel.EventLoopGroup;
+import io.opentelemetry.testing.internal.io.netty.channel.SimpleChannelInboundHandler;
+import io.opentelemetry.testing.internal.io.netty.channel.nio.NioEventLoopGroup;
+import io.opentelemetry.testing.internal.io.netty.channel.socket.SocketChannel;
+import io.opentelemetry.testing.internal.io.netty.channel.socket.nio.NioSocketChannel;
+import io.opentelemetry.testing.internal.io.netty.handler.codec.http.DefaultFullHttpRequest;
+import io.opentelemetry.testing.internal.io.netty.handler.codec.http.DefaultHttpResponse;
+import io.opentelemetry.testing.internal.io.netty.handler.codec.http.HttpClientCodec;
+import io.opentelemetry.testing.internal.io.netty.handler.codec.http.HttpObject;
+import io.opentelemetry.testing.internal.io.netty.handler.codec.http.HttpVersion;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Stream;
@@ -350,6 +369,88 @@ public abstract class AbstractHttpServerTest<SERVER> extends AbstractHttpServerU
     assertHighConcurrency(count);
   }
 
+  @Test
+  void httpPipelining() throws InterruptedException {
+    assumeTrue(options.testHttpPipelining);
+
+    int count = 10;
+    CountDownLatch countDownLatch = new CountDownLatch(count);
+    ServerEndpoint endpoint = INDEXED_CHILD;
+    TextMapPropagator propagator = GlobalOpenTelemetry.getPropagators().getTextMapPropagator();
+    TextMapSetter<DefaultFullHttpRequest> setter =
+        (request, key, value) -> request.headers().set(key, value);
+
+    EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
+    try {
+      Bootstrap bootstrap = buildBootstrap(eventLoopGroup);
+      Channel channel = bootstrap.connect(address.getHost(), port).sync().channel();
+      channel
+          .pipeline()
+          .addLast(
+              new SimpleChannelInboundHandler<HttpObject>() {
+
+                @Override
+                protected void channelRead0(
+                    ChannelHandlerContext channelHandlerContext, HttpObject httpObject) {
+                  if (httpObject instanceof DefaultHttpResponse) {
+                    countDownLatch.countDown();
+                  }
+                }
+              });
+
+      for (int i = 0; i < count; i++) {
+        int index = i;
+        String target =
+            endpoint.resolvePath(address).getPath().toString()
+                + "?"
+                + ServerEndpoint.ID_PARAMETER_NAME
+                + "="
+                + index;
+
+        testing.runWithSpan(
+            "client " + index,
+            () -> {
+              Span.current().setAttribute(ServerEndpoint.ID_ATTRIBUTE_NAME, index);
+              DefaultFullHttpRequest request =
+                  new DefaultFullHttpRequest(
+                      HttpVersion.HTTP_1_1,
+                      io.opentelemetry.testing.internal.io.netty.handler.codec.http.HttpMethod
+                          .valueOf("GET"),
+                      target,
+                      Unpooled.EMPTY_BUFFER);
+              request.headers().set(HttpHeaderNames.HOST, address.getHost() + ":" + port);
+              request.headers().set(HttpHeaderNames.USER_AGENT, TEST_USER_AGENT);
+              request.headers().set(HttpHeaderNames.X_FORWARDED_FOR, TEST_CLIENT_IP);
+
+              propagator.inject(Context.current(), request, setter);
+              channel.writeAndFlush(request);
+            });
+      }
+
+      countDownLatch.await(30, TimeUnit.SECONDS);
+      assertHighConcurrency(count);
+    } finally {
+      eventLoopGroup.shutdownGracefully().await(10, TimeUnit.SECONDS);
+    }
+  }
+
+  private static Bootstrap buildBootstrap(EventLoopGroup eventLoopGroup) {
+    Bootstrap bootstrap = new Bootstrap();
+    bootstrap
+        .group(eventLoopGroup)
+        .channel(NioSocketChannel.class)
+        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) TimeUnit.SECONDS.toMillis(10))
+        .handler(
+            new ChannelInitializer<SocketChannel>() {
+              @Override
+              protected void initChannel(SocketChannel socketChannel) {
+                ChannelPipeline pipeline = socketChannel.pipeline();
+                pipeline.addLast(new HttpClientCodec());
+              }
+            });
+    return bootstrap;
+  }
+
   protected void assertHighConcurrency(int count) {
     ServerEndpoint endpoint = INDEXED_CHILD;
     List<Consumer<TraceAssert>> assertions = new ArrayList<>();

testing-common/src/main/java/io/opentelemetry/instrumentation/testing/junit/http/HttpServerTestOptions.java

@@ -51,6 +51,7 @@ public final class HttpServerTestOptions {
   boolean testPathParam = false;
   boolean testCaptureHttpHeaders = true;
   boolean testCaptureRequestParameters = false;
+  boolean testHttpPipelining = true;
   boolean verifyServerSpanEndTime = true;
 
   HttpServerTestOptions() {}
@@ -182,6 +183,12 @@ public final class HttpServerTestOptions {
     return this;
   }
 
+  @CanIgnoreReturnValue
+  public HttpServerTestOptions setTestHttpPipelining(boolean testHttpPipelining) {
+    this.testHttpPipelining = testHttpPipelining;
+    return this;
+  }
+
   @CanIgnoreReturnValue
   public HttpServerTestOptions setVerifyServerSpanEndTime(boolean verifyServerSpanEndTime) {
     this.verifyServerSpanEndTime = verifyServerSpanEndTime;