Additional testing for Vert.x
Adds reactive/circuitbreaker tests.
ADD
dd-java-agent/instrumentation/vertx/src/test/groovy/VertxRxServerTest.groovy
+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
+ }
+}

ADD
dd-java-agent/instrumentation/vertx/src/test/groovy/VertxRxWebClientTest.groovy
+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
+ }
+}

ADD
dd-java-agent/instrumentation/vertx/src/test/java/VertxRxWebTestServer.java
+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() {
+ }
+}

MODIFY
dd-java-agent/instrumentation/vertx/vertx.gradle
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: '+'
}

Additional testing for Vert.x

Adds reactive/circuitbreaker tests.