Additional testing for Vert.x

Adds reactive/circuitbreaker tests.

dd-java-agent/instrumentation/vertx/src/test/groovy/VertxRxServerTest.groovy

@@ -0,0 +1,157 @@
+import datadog.trace.agent.test.AgentTestRunner
+import datadog.trace.agent.test.utils.OkHttpUtils
+import datadog.trace.agent.test.utils.PortUtils
+import datadog.trace.api.DDSpanTypes
+import io.netty.handler.codec.http.HttpResponseStatus
+import io.opentracing.tag.Tags
+import io.vertx.core.Vertx
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import spock.lang.Shared
+
+class VertxRxServerTest extends AgentTestRunner {
+
+  @Shared
+  OkHttpClient client = OkHttpUtils.client()
+
+  @Shared
+  int port
+  @Shared
+  Vertx server
+
+  def setupSpec() {
+    port = PortUtils.randomOpenPort()
+    server = VertxRxWebTestServer.start(port)
+  }
+
+  def cleanupSpec() {
+    server.close()
+  }
+
+  def "test server request/response"() {
+    setup:
+    def request = new Request.Builder()
+      .url("http://localhost:$port/proxy")
+      .header("x-datadog-trace-id", "123")
+      .header("x-datadog-parent-id", "456")
+      .get()
+      .build()
+    def response = client.newCall(request).execute()
+
+    expect:
+    response.code() == 200
+    response.body().string() == "Hello World"
+
+    and:
+    assertTraces(2) {
+      trace(0, 2) {
+        span(0) {
+          serviceName "unnamed-java-app"
+          operationName "netty.request"
+          resourceName "GET /test"
+          childOf(trace(1).get(1))
+          spanType DDSpanTypes.HTTP_SERVER
+          errored false
+          tags {
+            "$Tags.COMPONENT.key" "netty"
+            "$Tags.HTTP_METHOD.key" "GET"
+            "$Tags.HTTP_STATUS.key" 200
+            "$Tags.HTTP_URL.key" "http://localhost:$port/test"
+            "$Tags.PEER_HOSTNAME.key" "localhost"
+            "$Tags.PEER_HOST_IPV4.key" "127.0.0.1"
+            "$Tags.PEER_PORT.key" Integer
+            "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+            defaultTags(true)
+          }
+        }
+        span(1) {
+          childOf span(0)
+          assert span(1).operationName.endsWith('.tracedMethod')
+        }
+      }
+      trace(1, 2) {
+        span(0) {
+          serviceName "unnamed-java-app"
+          operationName "netty.request"
+          resourceName "GET /proxy"
+          traceId "123"
+          parentId "456"
+          spanType DDSpanTypes.HTTP_SERVER
+          errored false
+          tags {
+            "$Tags.COMPONENT.key" "netty"
+            "$Tags.HTTP_METHOD.key" "GET"
+            "$Tags.HTTP_STATUS.key" 200
+            "$Tags.HTTP_URL.key" "http://localhost:$port/proxy"
+            "$Tags.PEER_HOSTNAME.key" "localhost"
+            "$Tags.PEER_HOST_IPV4.key" "127.0.0.1"
+            "$Tags.PEER_PORT.key" Integer
+            "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+            defaultTags(true)
+          }
+        }
+        span(1) {
+          serviceName "unnamed-java-app"
+          operationName "netty.client.request"
+          resourceName "GET /test"
+          childOf(span(0))
+          spanType DDSpanTypes.HTTP_CLIENT
+          errored false
+          tags {
+            "$Tags.COMPONENT.key" "netty-client"
+            "$Tags.HTTP_METHOD.key" "GET"
+            "$Tags.HTTP_STATUS.key" 200
+            "$Tags.HTTP_URL.key" "http://localhost:$port/test"
+            "$Tags.PEER_HOSTNAME.key" "localhost"
+            "$Tags.PEER_HOST_IPV4.key" "127.0.0.1"
+            "$Tags.PEER_PORT.key" Integer
+            "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_CLIENT
+            defaultTags()
+          }
+        }
+      }
+    }
+  }
+
+  def "test #responseCode response handling"() {
+    setup:
+    def request = new Request.Builder().url("http://localhost:$port/$path").get().build()
+    def response = client.newCall(request).execute()
+
+    expect:
+    response.code() == responseCode.code()
+
+    and:
+    assertTraces(1) {
+      trace(0, 1) {
+        span(0) {
+          serviceName "unnamed-java-app"
+          operationName "netty.request"
+          resourceName name
+          spanType DDSpanTypes.HTTP_SERVER
+          errored error
+          tags {
+            "$Tags.COMPONENT.key" "netty"
+            "$Tags.HTTP_METHOD.key" "GET"
+            "$Tags.HTTP_STATUS.key" responseCode.code()
+            "$Tags.HTTP_URL.key" "http://localhost:$port/$path"
+            "$Tags.PEER_HOSTNAME.key" "localhost"
+            "$Tags.PEER_HOST_IPV4.key" "127.0.0.1"
+            "$Tags.PEER_PORT.key" Integer
+            "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+            if (error) {
+              tag("error", true)
+            }
+            defaultTags()
+          }
+        }
+      }
+    }
+
+    where:
+    responseCode                             | name         | path          | error
+    HttpResponseStatus.OK                    | "GET /"      | ""            | false
+    HttpResponseStatus.NOT_FOUND             | "404"        | "doesnt-exit" | false
+    HttpResponseStatus.INTERNAL_SERVER_ERROR | "GET /error" | "error"       | true
+  }
+}

dd-java-agent/instrumentation/vertx/src/test/groovy/VertxRxWebClientTest.groovy

@@ -0,0 +1,49 @@
+import datadog.trace.agent.test.base.HttpClientTest
+import datadog.trace.instrumentation.netty41.client.NettyHttpClientDecorator
+import io.vertx.core.VertxOptions
+import io.vertx.core.http.HttpMethod
+import io.vertx.reactivex.core.Vertx
+import io.vertx.reactivex.ext.web.client.WebClient
+import spock.lang.Shared
+import spock.lang.Timeout
+
+@Timeout(10)
+class VertxRxWebClientTest extends HttpClientTest<NettyHttpClientDecorator> {
+
+  @Shared
+  Vertx vertx = Vertx.vertx(new VertxOptions())
+  @Shared
+  WebClient client = WebClient.create(vertx);
+
+  @Override
+  int doRequest(String method, URI uri, Map<String, String> headers, Closure callback) {
+    def request = client.request(HttpMethod.valueOf(method), uri.port, uri.host, "$uri")
+    headers.each { request.putHeader(it.key, it.value) }
+    return request
+      .rxSend()
+      .doOnSuccess { response -> callback?.call() }
+      .map { it.statusCode() }
+      .toObservable()
+      .blockingFirst()
+  }
+
+  @Override
+  NettyHttpClientDecorator decorator() {
+    return NettyHttpClientDecorator.DECORATE
+  }
+
+  @Override
+  String expectedOperationName() {
+    return "netty.client.request"
+  }
+
+  @Override
+  boolean testRedirects() {
+    false
+  }
+
+  @Override
+  boolean testConnectionFailure() {
+    false
+  }
+}

dd-java-agent/instrumentation/vertx/src/test/java/VertxRxWebTestServer.java

@@ -0,0 +1,112 @@
+import datadog.trace.api.Trace;
+import io.vertx.circuitbreaker.CircuitBreakerOptions;
+import io.vertx.core.DeploymentOptions;
+import io.vertx.core.Future;
+import io.vertx.core.Vertx;
+import io.vertx.core.VertxOptions;
+import io.vertx.core.json.JsonObject;
+import io.vertx.reactivex.circuitbreaker.CircuitBreaker;
+import io.vertx.reactivex.core.AbstractVerticle;
+import io.vertx.reactivex.core.buffer.Buffer;
+import io.vertx.reactivex.ext.web.Router;
+import io.vertx.reactivex.ext.web.RoutingContext;
+import io.vertx.reactivex.ext.web.client.WebClient;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+public class VertxRxWebTestServer extends AbstractVerticle {
+  public static final String CONFIG_HTTP_SERVER_PORT = "http.server.port";
+
+  public static Vertx start(final int port) throws ExecutionException, InterruptedException {
+    /* This is highly against Vertx ideas, but our tests are synchronous
+    so we have to make sure server is up and running */
+    final CompletableFuture<Void> future = new CompletableFuture<>();
+
+    final Vertx vertx = Vertx.vertx(new VertxOptions().setClusterPort(port));
+
+    vertx.deployVerticle(
+      VertxRxWebTestServer.class.getName(),
+      new DeploymentOptions()
+        .setConfig(new JsonObject().put(CONFIG_HTTP_SERVER_PORT, port))
+        .setInstances(3),
+      res -> {
+        if (!res.succeeded()) {
+          throw new RuntimeException("Cannot deploy server Verticle", res.cause());
+        }
+        future.complete(null);
+      });
+
+    future.get();
+
+    return vertx;
+  }
+
+  @Override
+  public void start(final Future<Void> startFuture) {
+//    final io.vertx.reactivex.core.Vertx vertx = new io.vertx.reactivex.core.Vertx(this.vertx);
+    final WebClient client = WebClient.create(vertx);
+
+    final int port = config().getInteger(CONFIG_HTTP_SERVER_PORT);
+
+    final Router router = Router.router(vertx);
+    final CircuitBreaker breaker = CircuitBreaker.create("my-circuit-breaker", vertx,
+      new CircuitBreakerOptions()
+        .setMaxFailures(5) // number of failure before opening the circuit
+        .setTimeout(2000) // consider a failure if the operation does not succeed in time
+//        .setFallbackOnFailure(true) // do we call the fallback on failure
+        .setResetTimeout(10000) // time spent in open state before attempting to re-try
+    );
+
+    router
+      .route("/")
+      .handler(
+        routingContext -> {
+          routingContext.response().putHeader("content-type", "text/html").end("Hello World");
+        });
+    router
+      .route("/error")
+      .handler(
+        routingContext -> {
+          routingContext.response().setStatusCode(500).end();
+        });
+    router
+      .route("/proxy")
+      .handler(
+        routingContext -> {
+          breaker.execute(
+            ctx -> {
+              client.get(port, "localhost", "/test")
+                .rxSendBuffer(Optional.ofNullable(routingContext.getBody()).orElse(Buffer.buffer()))
+                .subscribe(
+                  response -> {
+                    routingContext
+                      .response()
+                      .setStatusCode(response.statusCode())
+                      .end(response.body());
+                  });
+            });
+        });
+    router
+      .route("/test")
+      .handler(
+        routingContext -> {
+          tracedMethod();
+          routingContext.next();
+        })
+      .blockingHandler(RoutingContext::next)
+      .handler(
+        routingContext -> {
+          routingContext.response().putHeader("content-type", "text/html").end("Hello World");
+        });
+
+    vertx
+      .createHttpServer()
+      .requestHandler(router::accept)
+      .listen(port, h -> startFuture.complete());
+  }
+
+  @Trace
+  private void tracedMethod() {
+  }
+}

dd-java-agent/instrumentation/vertx/vertx.gradle

@@ -48,6 +48,12 @@ dependencies {
 
   // Tests seem to fail before 3.5... maybe a problem with some of the tests?
   testCompile group: 'io.vertx', name: 'vertx-web', version: '3.5.0'
+  testCompile group: 'io.vertx', name: 'vertx-web-client', version: '3.5.0'
+  testCompile group: 'io.vertx', name: 'vertx-circuit-breaker', version: '3.5.0'
+  testCompile group: 'io.vertx', name: 'vertx-rx-java2', version: '3.5.0'
 
   latestDepTestCompile group: 'io.vertx', name: 'vertx-web', version: '+'
+  latestDepTestCompile group: 'io.vertx', name: 'vertx-web-client', version: '+'
+  latestDepTestCompile group: 'io.vertx', name: 'vertx-circuit-breaker', version: '+'
+  latestDepTestCompile group: 'io.vertx', name: 'vertx-rx-java2', version: '+'
 }