Test HTTP headers capturing on Netty-based server frameworks

instrumentation/finatra-2.9/javaagent/src/latestDepTest/groovy/FinatraServerLatestTest.groovy

@@ -94,8 +94,4 @@ class FinatraServerLatestTest extends HttpServerTest<HttpServer> implements Agen
       }
     }
   }
-
-  boolean testCapturedHttpHeaders() {
-    false
-  }
 }

instrumentation/finatra-2.9/javaagent/src/latestDepTest/scala/FinatraController.scala

@@ -65,4 +65,20 @@ class FinatraController extends Controller {
       }
     )
   }
+
+  any(CAPTURE_HEADERS.getPath) { request: Request =>
+    controller(
+      CAPTURE_HEADERS,
+      new Closure[Response](null) {
+        override def call(): Response = {
+          response
+            .ok(CAPTURE_HEADERS.getBody)
+            .header(
+              "X-Test-Response",
+              request.headerMap.get("X-Test-Request").get
+            )
+        }
+      }
+    )
+  }
 }

instrumentation/finatra-2.9/javaagent/src/test/groovy/FinatraServerTest.groovy

@@ -78,8 +78,4 @@ class FinatraServerTest extends HttpServerTest<HttpServer> implements AgentTestT
       }
     }
   }
-
-  boolean testCapturedHttpHeaders() {
-    false
-  }
 }

instrumentation/finatra-2.9/javaagent/src/test/scala/FinatraController.scala

@@ -66,6 +66,22 @@ class FinatraController extends Controller {
     )
   }
 
+  any(CAPTURE_HEADERS.getPath) { request: Request =>
+    controller(
+      CAPTURE_HEADERS,
+      new Closure[Response](null) {
+        override def call(): Response = {
+          response
+            .ok(CAPTURE_HEADERS.getBody)
+            .header(
+              "X-Test-Response",
+              request.headerMap.get("X-Test-Request").get
+            )
+        }
+      }
+    )
+  }
+
   any("/path/:id/param") { request: Request =>
     controller(
       PATH_PARAM,

instrumentation/netty/netty-3.8/javaagent/src/test/groovy/Netty38ServerTest.groovy

@@ -31,6 +31,7 @@ import org.jboss.netty.logging.InternalLoggerFactory
 import org.jboss.netty.logging.Slf4JLoggerFactory
 import org.jboss.netty.util.CharsetUtil
 
+import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.CAPTURE_HEADERS
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.INDEXED_CHILD
@@ -61,7 +62,8 @@ class Netty38ServerTest extends HttpServerTest<ServerBootstrap> implements Agent
       @Override
       void messageReceived(ChannelHandlerContext ctx, MessageEvent msg) throws Exception {
         if (msg.getMessage() instanceof HttpRequest) {
-          def uri = URI.create((msg.getMessage() as HttpRequest).getUri())
+          def request = msg.getMessage() as HttpRequest
+          def uri = URI.create(request.getUri())
           HttpServerTest.ServerEndpoint endpoint = forPath(uri.path)
           ctx.sendDownstream controller(endpoint) {
             HttpResponse response
@@ -90,6 +92,12 @@ class Netty38ServerTest extends HttpServerTest<ServerBootstrap> implements Agent
                 response.setContent(responseContent)
                 response.headers().set(LOCATION, endpoint.body)
                 break
+              case CAPTURE_HEADERS:
+                responseContent = ChannelBuffers.copiedBuffer(endpoint.body, CharsetUtil.UTF_8)
+                response = new DefaultHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status))
+                response.headers().set("X-Test-Response", request.headers().get("X-Test-Request"))
+                response.setContent(responseContent)
+                break
               case EXCEPTION:
                 throw new Exception(endpoint.body)
               default:
@@ -156,11 +164,6 @@ class Netty38ServerTest extends HttpServerTest<ServerBootstrap> implements Agent
     return "HTTP GET"
   }
 
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
-
   @Override
   boolean testConcurrency() {
     return true

instrumentation/netty/netty-4.0/javaagent/src/test/groovy/Netty40ServerTest.groovy

@@ -31,6 +31,7 @@ import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_LENGTH
 import static io.netty.handler.codec.http.HttpHeaders.Names.CONTENT_TYPE
 import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR
 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1
+import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.CAPTURE_HEADERS
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.INDEXED_CHILD
@@ -60,7 +61,8 @@ class Netty40ServerTest extends HttpServerTest<EventLoopGroup> implements AgentT
           pipeline.addLast([
             channelRead0       : { ctx, msg ->
               if (msg instanceof HttpRequest) {
-                def uri = URI.create((msg as HttpRequest).uri)
+                def request = msg as HttpRequest
+                def uri = URI.create(request.uri)
                 ServerEndpoint endpoint = ServerEndpoint.forPath(uri.path)
                 ctx.write controller(endpoint) {
                   ByteBuf content = null
@@ -85,6 +87,11 @@ class Netty40ServerTest extends HttpServerTest<EventLoopGroup> implements AgentT
                       response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status), content)
                       response.headers().set(HttpHeaders.Names.LOCATION, endpoint.body)
                       break
+                    case CAPTURE_HEADERS:
+                      content = Unpooled.copiedBuffer(endpoint.body, CharsetUtil.UTF_8)
+                      response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status), content)
+                      response.headers().set("X-Test-Response", request.headers().get("X-Test-Request"))
+                      break
                     case EXCEPTION:
                       throw new Exception(endpoint.body)
                     default:
@@ -126,11 +133,6 @@ class Netty40ServerTest extends HttpServerTest<EventLoopGroup> implements AgentT
     return "HTTP GET"
   }
 
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
-
   @Override
   boolean testConcurrency() {
     return true

instrumentation/netty/netty-4.1/javaagent/src/test/groovy/Netty41ServerTest.groovy

@@ -30,6 +30,7 @@ import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_LENGTH
 import static io.netty.handler.codec.http.HttpHeaderNames.CONTENT_TYPE
 import static io.netty.handler.codec.http.HttpResponseStatus.INTERNAL_SERVER_ERROR
 import static io.netty.handler.codec.http.HttpVersion.HTTP_1_1
+import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.CAPTURE_HEADERS
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.INDEXED_CHILD
@@ -59,7 +60,8 @@ class Netty41ServerTest extends HttpServerTest<EventLoopGroup> implements AgentT
           pipeline.addLast([
             channelRead0       : { ctx, msg ->
               if (msg instanceof HttpRequest) {
-                def uri = URI.create((msg as HttpRequest).uri())
+                def request = msg as HttpRequest
+                def uri = URI.create(request.uri())
                 ServerEndpoint endpoint = ServerEndpoint.forPath(uri.path)
                 ctx.write controller(endpoint) {
                   ByteBuf content = null
@@ -84,6 +86,11 @@ class Netty41ServerTest extends HttpServerTest<EventLoopGroup> implements AgentT
                       response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status), content)
                       response.headers().set(HttpHeaderNames.LOCATION, endpoint.body)
                       break
+                    case CAPTURE_HEADERS:
+                      content = Unpooled.copiedBuffer(endpoint.body, CharsetUtil.UTF_8)
+                      response = new DefaultFullHttpResponse(HTTP_1_1, HttpResponseStatus.valueOf(endpoint.status), content)
+                      response.headers().set("X-Test-Response", request.headers().get("X-Test-Request"))
+                      break
                     case EXCEPTION:
                       throw new Exception(endpoint.body)
                     default:
@@ -125,11 +132,6 @@ class Netty41ServerTest extends HttpServerTest<EventLoopGroup> implements AgentT
     return "HTTP GET"
   }
 
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
-
   @Override
   boolean testConcurrency() {
     return true

instrumentation/play/play-2.4/javaagent/src/test/groovy/server/PlayAsyncServerTest.groovy

@@ -13,6 +13,7 @@ import play.server.Server
 import java.util.concurrent.CompletableFuture
 import java.util.function.Supplier
 
+import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.CAPTURE_HEADERS
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.INDEXED_CHILD
@@ -54,6 +55,14 @@ class PlayAsyncServerTest extends PlayServerTest {
             Results.found(REDIRECT.getBody())
           }
         }, HttpExecution.defaultContext())
+      } as Supplier)
+        .GET(CAPTURE_HEADERS.getPath()).routeAsync({
+        CompletableFuture.supplyAsync({
+          controller(CAPTURE_HEADERS) {
+            Results.status(CAPTURE_HEADERS.getStatus(), CAPTURE_HEADERS.getBody())
+              .withHeader("X-Test-Response", request().getHeader("X-Test-Request"))
+          }
+        }, HttpExecution.defaultContext())
       } as Supplier)
         .GET(ERROR.getPath()).routeAsync({
         CompletableFuture.supplyAsync({

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

@@ -18,6 +18,7 @@ import play.server.Server
 import java.util.function.Supplier
 
 import static io.opentelemetry.api.trace.SpanKind.INTERNAL
+import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.CAPTURE_HEADERS
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.INDEXED_CHILD
@@ -51,6 +52,12 @@ class PlayServerTest extends HttpServerTest<Server> implements AgentTestTrait {
         controller(REDIRECT) {
           Results.found(REDIRECT.getBody())
         }
+      } as Supplier)
+        .GET(CAPTURE_HEADERS.getPath()).routeTo({
+        controller(CAPTURE_HEADERS) {
+          Results.status(CAPTURE_HEADERS.getStatus(), CAPTURE_HEADERS.getBody())
+            .withHeader("X-Test-Response", request().getHeader("X-Test-Request"))
+        }
       } as Supplier)
         .GET(ERROR.getPath()).routeTo({
         controller(ERROR) {
@@ -76,11 +83,6 @@ class PlayServerTest extends HttpServerTest<Server> implements AgentTestTrait {
     true
   }
 
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
-
   @Override
   boolean testConcurrency() {
     return true

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

@@ -13,9 +13,4 @@ class RatpackAsyncHttpServerTest extends AbstractRatpackAsyncHttpServerTest impl
   @Override
   void configure(RatpackServerSpec serverSpec) {
   }
-
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
 }

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

@@ -13,9 +13,4 @@ class RatpackForkedHttpServerTest extends AbstractRatpackForkedHttpServerTest im
   @Override
   void configure(RatpackServerSpec serverSpec) {
   }
-
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
 }

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

@@ -14,9 +14,4 @@ class RatpackHttpServerTest extends AbstractRatpackHttpServerTest implements Age
   @Override
   void configure(RatpackServerSpec serverSpec) {
   }
-
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
 }

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

@@ -61,11 +61,6 @@ abstract class SpringWebFluxServerTest extends HttpServerTest<ConfigurableApplic
     return true
   }
 
-  @Override
-  boolean testCapturedHttpHeaders() {
-    false
-  }
-
   @Override
   Class<?> expectedExceptionClass() {
     return IllegalStateException

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

@@ -101,6 +101,21 @@ Mono<String> indexed_child(ServerHttpRequest request, ServerHttpResponse respons
         });
   }
 
+  @GetMapping("/captureHeaders")
+  public Mono<String> capture_headers(ServerHttpRequest request, ServerHttpResponse response) {
+    ServerEndpoint endpoint = ServerEndpoint.CAPTURE_HEADERS;
+
+    return wrapControllerMethod(
+        endpoint,
+        () -> {
+          setStatus(response, endpoint);
+          response
+              .getHeaders()
+              .set("X-Test-Response", request.getHeaders().getFirst("X-Test-Request"));
+          return endpoint.getBody();
+        });
+  }
+
   protected abstract <T> Mono<T> wrapControllerMethod(ServerEndpoint endpoint, Callable<T> handler);
 
   private static void setStatus(ServerHttpResponse response, ServerEndpoint endpoint) {

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

@@ -84,6 +84,20 @@ public RouterFunction<ServerResponse> createRoutes() {
                       Span.current()
                           .setAttribute(
                               "test.request.id", Long.parseLong(request.queryParam("id").get())));
+            })
+        .andRoute(
+            GET("/captureHeaders"),
+            request -> {
+              ServerEndpoint endpoint = ServerEndpoint.CAPTURE_HEADERS;
+
+              return respond(
+                  endpoint,
+                  ServerResponse.status(endpoint.getStatus())
+                      .header(
+                          "X-Test-Response",
+                          request.headers().asHttpHeaders().getFirst("X-Test-Request")),
+                  null,
+                  null);
             });
   }
 

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

@@ -12,6 +12,7 @@ import io.vertx.reactivex.circuitbreaker.CircuitBreaker
 import io.vertx.reactivex.core.AbstractVerticle
 import io.vertx.reactivex.ext.web.Router
 
+import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.CAPTURE_HEADERS
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.ERROR
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.EXCEPTION
 import static io.opentelemetry.instrumentation.test.base.HttpServerTest.ServerEndpoint.INDEXED_CHILD
@@ -134,6 +135,22 @@ class VertxRxCircuitBreakerHttpServerTest extends VertxRxHttpServerTest {
           }
         })
       }
+      router.route(CAPTURE_HEADERS.path).handler { ctx ->
+        breaker.execute({ future ->
+          future.complete(CAPTURE_HEADERS)
+        }, {
+          if (it.failed()) {
+            throw it.cause()
+          }
+          HttpServerTest.ServerEndpoint endpoint = it.result()
+          controller(endpoint) {
+            ctx.response().setStatusCode(endpoint.status)
+              .putHeader("X-Test-Response", ctx.request().getHeader("X-Test-Request"))
+              .end(endpoint.body)
+          }
+        })
+      }
+
 
       super.@vertx.createHttpServer()
         .requestHandler(router)