Spring webflux instrumentation
instruments spring webflux 5+
ADD
dd-java-agent/instrumentation/spring-webflux/spring-webflux.gradle
+muzzle {
+ pass {
+ group = "org.springframework"
+ module = "spring-webflux"
+ versions = "[5.0.0.RELEASE,)"
+ assertInverse = true
+ }
+}
+
+apply from: "${rootDir}/gradle/java.gradle"
+
+sourceSets {
+ main_java8 {
+ java.srcDirs "${project.projectDir}/src/main/java8"
+ }
+}
+
+compileMain_java8Java {
+ sourceCompatibility = 1.8
+ targetCompatibility = 1.8
+}
+
+apply plugin: 'org.unbroken-dome.test-sets'
+
+testSets {
+ latestDepTest {
+ dirName = 'test'
+ }
+}
+
+dependencies {
+
+ main_java8CompileOnly group: 'org.springframework', name: 'spring-webflux', version: '5.0.0.RELEASE'
+
+ main_java8Compile project(':dd-java-agent:agent-tooling')
+
+ main_java8Compile deps.bytebuddy
+ main_java8Compile deps.opentracing
+
+ compileOnly sourceSets.main_java8.compileClasspath
+
+ compile sourceSets.main_java8.output
+
+ compileOnly group: 'org.springframework', name: 'spring-webflux', version: '5.0.0.RELEASE'
+
+ compile project(':dd-java-agent:agent-tooling')
+
+ compile deps.bytebuddy
+ compile deps.opentracing
+ annotationProcessor deps.autoservice
+ implementation deps.autoservice
+
+ testCompile project(':dd-java-agent:testing')
+ testCompile project(':dd-java-agent:instrumentation:netty-4.1')
+ testCompile project(':dd-java-agent:instrumentation:java-concurrent')
+
+ testCompile group: 'org.springframework.boot', name: 'spring-boot-starter-webflux', version: '2.0.0.RELEASE'
+ testCompile group: 'org.springframework.boot', name: 'spring-boot-starter', version: '2.0.0.RELEASE'
+ testCompile group: 'org.spockframework', name: 'spock-spring', version: '1.1-groovy-2.4'
+ testCompile group: 'org.springframework.boot', name: 'spring-boot-starter-test', version: '2.0.0.RELEASE'
+ testCompile group: 'org.springframework.boot', name: 'spring-boot-starter-reactor-netty', version: '2.0.0.RELEASE'
+}
+
+configurations.latestDepTestCompile {
+ resolutionStrategy {
+ force group: 'org.springframework.boot', name: 'spring-boot-starter-webflux', version: '+'
+ force group: 'org.springframework.boot', name: 'spring-boot-starter-test', version: '+'
+ force group: 'org.springframework.boot', name: 'spring-boot-starter-reactor-netty', version: '+'
+ }
+}
+
+testJava8Only += '**/*Test.class'
+testJava8Only += '**/*TestApplication.class'
+testJava8Only += '**/EchoHandler.class'
+testJava8Only += '**/EchoHandlerFunction.class'
+testJava8Only += '**/FooModel.class'
+testJava8Only += '**/TestController.class'

ADD
dd-java-agent/instrumentation/spring-webflux/src/main/java/datadog/trace/instrumentation/springwebflux/DispatcherHandlerInstrumentation.java
+package datadog.trace.instrumentation.springwebflux;
+
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.Instrumenter;
+import java.util.Collections;
+import java.util.Map;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+@AutoService(Instrumenter.class)
+public final class DispatcherHandlerInstrumentation extends Instrumenter.Default {
+
+ public static final String PACKAGE =
+ DispatcherHandlerInstrumentation.class.getPackage().getName();
+
+ public DispatcherHandlerInstrumentation() {
+ super("spring-webflux");
+ }
+
+ @Override
+ public String[] helperClassNames() {
+ return new String[] {PACKAGE + ".DispatcherHandlerMonoBiConsumer"};
+ }
+
+ @Override
+ public ElementMatcher<TypeDescription> typeMatcher() {
+ return named("org.springframework.web.reactive.DispatcherHandler");
+ }
+
+ @Override
+ public Map<ElementMatcher, String> transformers() {
+ return Collections.<ElementMatcher, String>singletonMap(
+ isMethod()
+ .and(isPublic())
+ .and(named("handle"))
+ .and(takesArgument(0, named("org.springframework.web.server.ServerWebExchange")))
+ .and(takesArguments(1)),
+ // Cannot reference class directly here because it would lead to class load failure on Java7
+ PACKAGE + ".DispatcherHandlerAdvice");
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/main/java/datadog/trace/instrumentation/springwebflux/HandlerFunctionAdapterInstrumentation.java
+package datadog.trace.instrumentation.springwebflux;
+
+import static io.opentracing.log.Fields.ERROR_OBJECT;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.Instrumenter;
+import io.opentracing.Span;
+import io.opentracing.tag.Tags;
+import io.opentracing.util.GlobalTracer;
+import java.util.Collections;
+import java.util.Map;
+import net.bytebuddy.asm.Advice;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+@AutoService(Instrumenter.class)
+public final class HandlerFunctionAdapterInstrumentation extends Instrumenter.Default {
+
+ public HandlerFunctionAdapterInstrumentation() {
+ super("spring-webflux");
+ }
+
+ @Override
+ public ElementMatcher<TypeDescription> typeMatcher() {
+ return named("org.springframework.web.reactive.function.server.support.HandlerFunctionAdapter");
+ }
+
+ @Override
+ public Map<ElementMatcher, String> transformers() {
+ return Collections.<ElementMatcher, String>singletonMap(
+ isMethod()
+ .and(isPublic())
+ .and(named("handle"))
+ .and(takesArgument(0, named("org.springframework.web.server.ServerWebExchange")))
+ .and(takesArguments(2)),
+ HandlerFunctionAdapterAdvice.class.getName());
+ }
+
+ public static class HandlerFunctionAdapterAdvice {
+
+ @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+ public static void recordHandlerFunctionTag(
+ @Advice.Thrown final Throwable throwable, @Advice.Argument(1) final Object handler) {
+
+ final Span activeSpan = GlobalTracer.get().activeSpan();
+
+ if (activeSpan != null && handler != null) {
+ activeSpan.setTag("handler_function.class", handler.getClass().getName());
+ if (throwable != null) {
+ Tags.ERROR.set(activeSpan, true);
+ activeSpan.log(Collections.singletonMap(ERROR_OBJECT, throwable));
+ }
+ }
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/main/java/datadog/trace/instrumentation/springwebflux/RouterFunctionInstrumentation.java
+package datadog.trace.instrumentation.springwebflux;
+
+import static datadog.trace.agent.tooling.ByteBuddyElementMatchers.safeHasSuperType;
+import static net.bytebuddy.matcher.ElementMatchers.declaresField;
+import static net.bytebuddy.matcher.ElementMatchers.isAbstract;
+import static net.bytebuddy.matcher.ElementMatchers.isMethod;
+import static net.bytebuddy.matcher.ElementMatchers.isPublic;
+import static net.bytebuddy.matcher.ElementMatchers.named;
+import static net.bytebuddy.matcher.ElementMatchers.not;
+import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
+import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
+
+import com.google.auto.service.AutoService;
+import datadog.trace.agent.tooling.Instrumenter;
+import java.util.Collections;
+import java.util.Map;
+import net.bytebuddy.description.type.TypeDescription;
+import net.bytebuddy.matcher.ElementMatcher;
+
+@AutoService(Instrumenter.class)
+public final class RouterFunctionInstrumentation extends Instrumenter.Default {
+
+ public static final String PACKAGE = RouterFunctionInstrumentation.class.getPackage().getName();
+
+ public RouterFunctionInstrumentation() {
+ super("spring-webflux");
+ }
+
+ @Override
+ public String[] helperClassNames() {
+ return new String[] {
+ RouterFunctionInstrumentation.class.getPackage().getName()
+ + ".DispatcherHandlerMonoBiConsumer"
+ };
+ }
+
+ @Override
+ public ElementMatcher<TypeDescription> typeMatcher() {
+ return not(isAbstract())
+ .and(declaresField(named("predicate")))
+ .and(
+ safeHasSuperType(
+ named(
+ "org.springframework.web.reactive.function.server.RouterFunctions$AbstractRouterFunction")));
+ }
+
+ @Override
+ public Map<ElementMatcher, String> transformers() {
+ return Collections.<ElementMatcher, String>singletonMap(
+ isMethod()
+ .and(isPublic())
+ .and(named("route"))
+ .and(
+ takesArgument(
+ 0, named("org.springframework.web.reactive.function.server.ServerRequest")))
+ .and(takesArguments(1)),
+ // Cannot reference class directly here because it would lead to class load failure on Java7
+ PACKAGE + ".RouterFunctionAdvice");
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/main/java8/datadog/trace/instrumentation/springwebflux/DispatcherHandlerAdvice.java
+package datadog.trace.instrumentation.springwebflux;
+
+import static io.opentracing.log.Fields.ERROR_OBJECT;
+
+import io.opentracing.Scope;
+import io.opentracing.tag.Tags;
+import io.opentracing.util.GlobalTracer;
+import java.util.Collections;
+import net.bytebuddy.asm.Advice;
+import reactor.core.publisher.Mono;
+
+public class DispatcherHandlerAdvice {
+
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static Scope startSpan() {
+ return GlobalTracer.get()
+ .buildSpan("DispatcherHandler.handle")
+ .withTag(Tags.COMPONENT.getKey(), "spring-webflux-controller")
+ .startActive(true);
+ }
+
+ @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+ public static void addBehaviorTrigger(
+ @Advice.Enter final Scope scope,
+ @Advice.Thrown final Throwable throwable,
+ @Advice.Return(readOnly = false) Mono<Void> returnMono) {
+
+ if (scope != null) {
+ if (throwable != null) {
+ Tags.ERROR.set(scope.span(), true);
+ scope.span().log(Collections.singletonMap(ERROR_OBJECT, throwable));
+ } else {
+ returnMono = returnMono.doOnSuccessOrError(new DispatcherHandlerMonoBiConsumer<>(scope));
+ }
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/main/java8/datadog/trace/instrumentation/springwebflux/DispatcherHandlerMonoBiConsumer.java
+package datadog.trace.instrumentation.springwebflux;
+
+import static io.opentracing.log.Fields.ERROR_OBJECT;
+
+import datadog.trace.api.DDSpanTypes;
+import datadog.trace.api.DDTags;
+import io.opentracing.Scope;
+import io.opentracing.Span;
+import io.opentracing.tag.Tags;
+import io.opentracing.util.GlobalTracer;
+import java.util.Collections;
+import java.util.function.BiConsumer;
+
+public class DispatcherHandlerMonoBiConsumer<U> implements BiConsumer<U, Throwable> {
+
+ private final Scope scope;
+ public static final ThreadLocal<String> tlsPathUrlTag = new ThreadLocal<>();
+
+ public DispatcherHandlerMonoBiConsumer(final Scope scope) {
+ this.scope = scope;
+ }
+
+ @Override
+ public void accept(final U object, final Throwable throwable) {
+ final Span spanToChange = scope.span();
+ if (throwable != null) {
+ spanToChange.log(Collections.singletonMap(ERROR_OBJECT, throwable));
+ Tags.ERROR.set(spanToChange, true);
+ }
+
+ scope.close();
+
+ final Span parentSpan = GlobalTracer.get().activeSpan();
+ final String pathUrl = tlsPathUrlTag.get();
+
+ if (pathUrl != null && parentSpan != null) {
+ parentSpan.setTag(DDTags.RESOURCE_NAME, pathUrl);
+ parentSpan.setTag(DDTags.SPAN_TYPE, DDSpanTypes.WEB_SERVLET);
+ tlsPathUrlTag.remove();
+ }
+ }
+
+ public static void setTLPathUrl(final String pathUrl) {
+ tlsPathUrlTag.set(pathUrl);
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/main/java8/datadog/trace/instrumentation/springwebflux/RouterFunctionAdvice.java
+package datadog.trace.instrumentation.springwebflux;
+
+import static io.opentracing.log.Fields.ERROR_OBJECT;
+
+import datadog.trace.api.DDSpanTypes;
+import datadog.trace.api.DDTags;
+import io.opentracing.Scope;
+import io.opentracing.Span;
+import io.opentracing.tag.Tags;
+import io.opentracing.util.GlobalTracer;
+import java.util.Collections;
+import net.bytebuddy.asm.Advice;
+import org.springframework.web.reactive.function.server.RequestPredicate;
+import org.springframework.web.reactive.function.server.RequestPredicates;
+import org.springframework.web.reactive.function.server.ServerRequest;
+
+public class RouterFunctionAdvice {
+
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static void nameResource(
+ @Advice.FieldValue(value = "predicate") final RequestPredicate predicate,
+ @Advice.Argument(0) final ServerRequest serverRequest) {
+
+ if (predicate == null) {
+ return;
+ }
+
+ final Class predicateEnclosingClass = predicate.getClass().getEnclosingClass();
+ final String predicateString = predicate.toString();
+ if (predicate.test(serverRequest)
+ && serverRequest != null
+ && predicateString != null
+ && !predicateString.isEmpty()
+ && predicateEnclosingClass == RequestPredicates.class) {
+
+ // only change parent span if the predicate is one of those enclosed in
+ // org.springframework.web.reactive.function.server RequestPredicates
+ // otherwise the parent may have weird resource names such as lambda request predicate class
+ // names that arise from webflux error handling
+
+ final String resourceName =
+ predicateString
+ .replace("&&", "")
+ .replace("||", "")
+ .replace("(", "")
+ .replace(")", "")
+ .replaceAll("[ \\t]+", " ");
+
+ // to be used as resource name by netty span, most likely
+ DispatcherHandlerMonoBiConsumer.setTLPathUrl(resourceName);
+
+ // should be the dispatcher handler span
+ final Scope activeScope = GlobalTracer.get().scopeManager().active();
+ if (activeScope != null) {
+ activeScope.span().setTag("predicate_string", predicateString);
+ activeScope.span().setTag(DDTags.SPAN_TYPE, DDSpanTypes.HTTP_SERVER);
+ }
+ }
+ }
+
+ @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
+ public static void recordThrowable(@Advice.Thrown final Throwable throwable) {
+ final Scope scope = GlobalTracer.get().scopeManager().active();
+ if (scope != null && throwable != null) {
+ final Span span = scope.span();
+ Tags.ERROR.set(span, true);
+ span.log(Collections.singletonMap(ERROR_OBJECT, throwable));
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/test/groovy/SpringWebfluxTest.groovy
+import datadog.trace.agent.test.AgentTestRunner
+import datadog.trace.agent.test.utils.OkHttpUtils
+import datadog.trace.api.DDSpanTypes
+import datadog.trace.api.DDTags
+import datadog.trace.instrumentation.springwebflux.EchoHandler
+import datadog.trace.instrumentation.springwebflux.FooModel
+import datadog.trace.instrumentation.springwebflux.SpringWebFluxTestApplication
+import io.opentracing.tag.Tags
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.RequestBody
+import org.springframework.boot.test.context.SpringBootTest
+import org.springframework.boot.web.server.LocalServerPort
+import org.springframework.web.server.ResponseStatusException
+import spock.lang.Unroll
+
+import static datadog.trace.agent.test.asserts.ListWriterAssert.assertTraces
+
+@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT, classes = SpringWebFluxTestApplication)
+class SpringWebfluxTest extends AgentTestRunner {
+
+ static final okhttp3.MediaType PLAIN_TYPE = okhttp3.MediaType.parse("text/plain; charset=utf-8")
+ static final String INNER_HANDLER_FUNCTION_CLASS_TAG_PREFIX = SpringWebFluxTestApplication.getName() + "\$"
+
+ @LocalServerPort
+ private int port
+
+ OkHttpClient client = OkHttpUtils.client()
+
+ @Unroll
+ def "Basic GET test #testName"() {
+ setup:
+ String url = "http://localhost:$port/greet$urlSuffix"
+ def request = new Request.Builder().url(url).get().build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code == 200
+ response.body().string() == expectedResponseBody
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(1))
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "predicate_string" "(GET && /greet$pathVariableUrlSuffix)"
+ "handler_function.class" { String tagVal ->
+ return tagVal.contains(INNER_HANDLER_FUNCTION_CLASS_TAG_PREFIX)
+ }
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "GET /greet$pathVariableUrlSuffix"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 200
+ "$Tags.HTTP_URL.key" url
+ defaultTags()
+ }
+ }
+ }
+ }
+
+ where:
+ testName | urlSuffix | pathVariableUrlSuffix | expectedResponseBody
+ "without paramaters" | "" | "" | SpringWebFluxTestApplication.GreetingHandler.DEFAULT_RESPONSE
+ "with one parameter" | "/WORLD" | "/{name}" | SpringWebFluxTestApplication.GreetingHandler.DEFAULT_RESPONSE + " WORLD"
+ "with 2 parameters" | "/World/Test1" | "/{name}/{word}" | SpringWebFluxTestApplication.GreetingHandler.DEFAULT_RESPONSE + " World Test1"
+ }
+
+ def "404 GET test"() {
+ setup:
+ String url = "http://localhost:$port/notfoundgreet"
+ def request = new Request.Builder().url(url).get().build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code == 404
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "404"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 404
+ "$Tags.HTTP_URL.key" url
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(0))
+ errored true
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ errorTags(ResponseStatusException, String)
+ defaultTags()
+ }
+ }
+ }
+ }
+ }
+
+ def "Basic POST test"() {
+ setup:
+ String echoString = "TEST"
+ String url = "http://localhost:$port/echo"
+ RequestBody body = RequestBody.create(PLAIN_TYPE, echoString)
+ def request = new Request.Builder().url(url).post(body).build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code() == 202
+ response.body().string() == echoString
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(1))
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "predicate_string" "(POST && /echo)"
+ "handler_function.class" { String tagVal ->
+ return tagVal.contains(EchoHandler.getName())
+ }
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "POST /echo"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "POST"
+ "$Tags.HTTP_STATUS.key" 202
+ "$Tags.HTTP_URL.key" url
+ defaultTags()
+ }
+ }
+ }
+ }
+ }
+
+ def "GET to bad annotation API endpoint"() {
+ setup:
+ String url = "http://localhost:$port/failfoo/1"
+ def request = new Request.Builder().url(url).get().build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code() == 500
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "GET /failfoo/?"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ errored true
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 500
+ "$Tags.HTTP_URL.key" url
+ "error" true
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(0))
+ errored true
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ errorTags(ArithmeticException, String)
+ defaultTags()
+ }
+ }
+ }
+ }
+ }
+
+ def "POST to bad functional API endpoint"() {
+ setup:
+ String echoString = "TEST"
+ RequestBody body = RequestBody.create(PLAIN_TYPE, echoString)
+ String url = "http://localhost:$port/fail-echo"
+ def request = new Request.Builder().url(url).post(body).build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code() == 500
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "POST /fail-echo"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ errored true
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "POST"
+ "$Tags.HTTP_STATUS.key" 500
+ "$Tags.HTTP_URL.key" url
+ "error" true
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(0))
+ errored true
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "predicate_string" "(POST && /fail-echo)"
+ "handler_function.class" { String tagVal ->
+ return tagVal.contains(INNER_HANDLER_FUNCTION_CLASS_TAG_PREFIX)
+ }
+ errorTags(NullPointerException, String)
+ defaultTags()
+ }
+ }
+ }
+ }
+ }
+
+ def "Redirect test"() {
+ setup:
+ String url = "http://localhost:$port/double-greet-redirect"
+ def request = new Request.Builder().url(url).get().build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code == 200
+ assertTraces(TEST_WRITER, 2) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "GET /double-greet-redirect"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 307
+ "$Tags.HTTP_URL.key" url
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(0))
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "predicate_string" "(GET && /double-greet-redirect)"
+ "handler_function.class" { String tagVal -> // lambda function, loosen restriction on this tag
+ return (tagVal.contains(INNER_HANDLER_FUNCTION_CLASS_TAG_PREFIX)
+ || tagVal.contains("Lambda") || tagVal.contains("Proxy"))
+ }
+ defaultTags()
+ }
+ }
+ }
+ trace(1, 2) {
+ span(0) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(1))
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "predicate_string" "(GET && /double-greet)"
+ "handler_function.class" { String tagVal ->
+ return tagVal.contains(INNER_HANDLER_FUNCTION_CLASS_TAG_PREFIX)
+ }
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "GET /double-greet"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 200
+ "$Tags.HTTP_URL.key" "http://localhost:$port/double-greet"
+ defaultTags()
+ }
+ }
+ }
+ }
+ }
+
+ @Unroll
+ def "Flux x#count GET test with functional API endpoint"() {
+ setup:
+ String expectedResponseBodyStr = FooModel.createXFooModelsStringFromArray(FooModel.createXFooModels(count))
+ String url = "http://localhost:$port/greet-counter/$count"
+ def request = new Request.Builder().url(url).get().build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code() == 200
+ expectedResponseBodyStr == response.body().string()
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(1))
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "predicate_string" "(GET && /greet-counter/{count})"
+ "handler_function.class" { String tagVal ->
+ return tagVal.contains(INNER_HANDLER_FUNCTION_CLASS_TAG_PREFIX)
+ }
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "GET /greet-counter/{count}"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 200
+ "$Tags.HTTP_URL.key" url
+ defaultTags()
+ }
+ }
+ }
+ }
+
+ where:
+ count << [0, 1, 10]
+ }
+
+ @Unroll
+ def "Flux x#count GET test with spring annotations endpoint"() {
+ setup:
+ String expectedResponseBodyStr = FooModel.createXFooModelsStringFromArray(FooModel.createXFooModels(count))
+ String url = "http://localhost:$port/annotation-foos/$count"
+ def request = new Request.Builder().url(url).get().build()
+
+ when:
+ def response = client.newCall(request).execute()
+
+ then:
+ response.code() == 200
+ expectedResponseBodyStr == response.body().string()
+ assertTraces(TEST_WRITER, 1) {
+ trace(0, 2) {
+ span(0) {
+ resourceName "DispatcherHandler.handle"
+ operationName "DispatcherHandler.handle"
+ spanType DDSpanTypes.HTTP_SERVER
+ childOf(span(1))
+ tags {
+ "$Tags.COMPONENT.key" "spring-webflux-controller"
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ defaultTags()
+ }
+ }
+ span(1) {
+ resourceName "GET /annotation-foos/?"
+ operationName "netty.request"
+ spanType DDSpanTypes.HTTP_SERVER
+ parent()
+ tags {
+ "$Tags.COMPONENT.key" "netty"
+ "$Tags.SPAN_KIND.key" Tags.SPAN_KIND_SERVER
+ "$DDTags.SPAN_TYPE" DDSpanTypes.HTTP_SERVER
+ "$Tags.PEER_HOSTNAME.key" "localhost"
+ "$Tags.PEER_PORT.key" Integer
+ "$Tags.HTTP_METHOD.key" "GET"
+ "$Tags.HTTP_STATUS.key" 200
+ "$Tags.HTTP_URL.key" url
+ defaultTags()
+ }
+ }
+ }
+ }
+
+ where:
+ count << [0, 1, 10]
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/test/groovy/datadog/trace/instrumentation/springwebflux/EchoHandler.groovy
+package datadog.trace.instrumentation.springwebflux
+
+import org.springframework.http.MediaType
+import org.springframework.stereotype.Component
+import org.springframework.web.reactive.function.server.ServerRequest
+import org.springframework.web.reactive.function.server.ServerResponse
+import reactor.core.publisher.Mono
+
+@Component
+class EchoHandler {
+ Mono<ServerResponse> echo(ServerRequest request) {
+ return ServerResponse.accepted().contentType(MediaType.TEXT_PLAIN)
+ .body(request.bodyToMono(String), String)
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/test/groovy/datadog/trace/instrumentation/springwebflux/EchoHandlerFunction.groovy
+package datadog.trace.instrumentation.springwebflux
+
+import org.springframework.web.reactive.function.server.HandlerFunction
+import org.springframework.web.reactive.function.server.ServerRequest
+import org.springframework.web.reactive.function.server.ServerResponse
+import reactor.core.publisher.Mono
+
+class EchoHandlerFunction implements HandlerFunction<ServerResponse> {
+
+ EchoHandler echoHandler
+
+ EchoHandlerFunction(EchoHandler echoHandler) {
+ this.echoHandler = echoHandler
+ }
+
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return echoHandler.echo(request)
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/test/groovy/datadog/trace/instrumentation/springwebflux/FooModel.groovy
+package datadog.trace.instrumentation.springwebflux
+
+class FooModel {
+ public long id
+ public String name
+
+ FooModel(long id, String name) {
+ this.id = id
+ this.name = name
+ }
+
+ @Override
+ String toString() {
+ return "{\"id\":" + id + ",\"name\":\"" + name + "\"}"
+ }
+
+ static FooModel[] createXFooModels(long count) {
+ FooModel[] foos = new FooModel[count]
+ for (int i = 0; i < count; ++i) {
+ foos[i] = new FooModel(i, String.valueOf(i))
+ }
+ return foos
+ }
+
+ static String createXFooModelsStringFromArray(FooModel[] foos) {
+ StringBuilder sb = new StringBuilder()
+ sb.append("[")
+ for (int i = 0; i < foos.length; ++i) {
+ sb.append(foos[i].toString())
+ if (i < foos.length - 1) {
+ sb.append(",")
+ }
+ }
+ sb.append("]")
+ return sb.toString()
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/test/groovy/datadog/trace/instrumentation/springwebflux/SpringWebFluxTestApplication.groovy
+package datadog.trace.instrumentation.springwebflux
+
+import org.springframework.boot.SpringApplication
+import org.springframework.boot.WebApplicationType
+import org.springframework.boot.autoconfigure.AutoConfigureBefore
+import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.autoconfigure.web.reactive.ReactiveWebServerFactoryAutoConfiguration
+import org.springframework.boot.test.context.TestConfiguration
+import org.springframework.boot.web.embedded.netty.NettyReactiveWebServerFactory
+import org.springframework.context.annotation.Bean
+import org.springframework.http.MediaType
+import org.springframework.stereotype.Component
+import org.springframework.web.reactive.function.BodyInserters
+import org.springframework.web.reactive.function.server.HandlerFunction
+import org.springframework.web.reactive.function.server.RouterFunction
+import org.springframework.web.reactive.function.server.ServerRequest
+import org.springframework.web.reactive.function.server.ServerResponse
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+import static org.springframework.web.reactive.function.server.RequestPredicates.GET
+import static org.springframework.web.reactive.function.server.RequestPredicates.POST
+import static org.springframework.web.reactive.function.server.RouterFunctions.route
+
+@SpringBootApplication
+class SpringWebFluxTestApplication {
+
+ static void main(String[] args) {
+ SpringApplication app = new SpringApplication(SpringWebFluxTestApplication)
+ app.setWebApplicationType(WebApplicationType.REACTIVE)
+ app.run(args)
+ }
+
+ @TestConfiguration
+ @AutoConfigureBefore(ReactiveWebServerFactoryAutoConfiguration)
+ class ForceNettyAutoConfiguration {
+ @Bean
+ NettyReactiveWebServerFactory nettyFactory() {
+ return new NettyReactiveWebServerFactory()
+ }
+ }
+
+ @Bean
+ RouterFunction<ServerResponse> redirectRouterFunction() {
+ return route(GET("/double-greet-redirect"), {
+ req -> ServerResponse.temporaryRedirect(URI.create("/double-greet")).build()
+ })
+ }
+
+ @Bean
+ RouterFunction<ServerResponse> echoRouterFunction(EchoHandler echoHandler) {
+ return route(POST("/echo"), new EchoHandlerFunction(echoHandler))
+ .andRoute(POST("/fail-echo"), new HandlerFunction<ServerResponse>() {
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return echoHandler.echo(null)
+ }
+ })
+ }
+
+ @Bean
+ RouterFunction<ServerResponse> greetRouterFunction(GreetingHandler greetingHandler) {
+ return route(GET("/greet"), new HandlerFunction<ServerResponse>() {
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return greetingHandler.defaultGreet()
+ }
+ }).andRoute(GET("/greet/{name}"), new HandlerFunction<ServerResponse>() {
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return greetingHandler.customGreet(request)
+ }
+ }).andRoute(GET("/greet/{name}/{word}"), new HandlerFunction<ServerResponse>() {
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return greetingHandler.customGreetWithWord(request)
+ }
+ }).andRoute(GET("/double-greet"), new HandlerFunction<ServerResponse>() {
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return greetingHandler.doubleGreet()
+ }
+ }).andRoute(GET("/greet-counter/{count}"), new HandlerFunction<ServerResponse>() {
+ @Override
+ Mono<ServerResponse> handle(ServerRequest request) {
+ return greetingHandler.counterGreet(request)
+ }
+ })
+ }
+
+ @Component
+ class GreetingHandler {
+ static final String DEFAULT_RESPONSE = "HELLO"
+
+ Mono<ServerResponse> defaultGreet() {
+ return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
+ .body(BodyInserters.fromObject(DEFAULT_RESPONSE))
+ }
+
+ Mono<ServerResponse> doubleGreet() {
+ return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
+ .body(BodyInserters.fromObject(DEFAULT_RESPONSE + DEFAULT_RESPONSE))
+ }
+
+ Mono<ServerResponse> customGreet(ServerRequest request) {
+ return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
+ .body(BodyInserters.fromObject(DEFAULT_RESPONSE + " " + request.pathVariable("name")))
+ }
+
+ Mono<ServerResponse> customGreetWithWord(ServerRequest request) {
+ return ServerResponse.ok().contentType(MediaType.TEXT_PLAIN)
+ .body(BodyInserters.fromObject(DEFAULT_RESPONSE + " " + request.pathVariable("name") + " " + request.pathVariable("word")))
+ }
+
+ Mono<ServerResponse> counterGreet(ServerRequest request) {
+ final int countTo = Integer.valueOf(request.pathVariable("count"))
+ FooModel[] fooArray = FooModel.createXFooModels(countTo)
+ return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
+ .body(Flux.fromArray(fooArray), FooModel)
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/spring-webflux/src/test/groovy/datadog/trace/instrumentation/springwebflux/TestController.groovy
+package datadog.trace.instrumentation.springwebflux
+
+
+import org.springframework.web.bind.annotation.GetMapping
+import org.springframework.web.bind.annotation.PathVariable
+import org.springframework.web.bind.annotation.RestController
+import reactor.core.publisher.Flux
+import reactor.core.publisher.Mono
+
+@RestController
+class TestController {
+
+ @GetMapping("/failfoo/{id}")
+ Mono<FooModel> getFooModelFail(@PathVariable("id") long id) {
+ return Mono.just(new FooModel((id / 0), "fail"))
+ }
+
+ @GetMapping(value = "/annotation-foos/{count}")
+ Flux<FooModel> getXFooModelsWithDelay(@PathVariable("count") long count) {
+ FooModel[] foos = FooModel.createXFooModels(count)
+ return Flux.just(foos)
+ }
+}

MODIFY
settings.gradle
settings.gradle
@@ -48,6 +48,7 @@ include ':dd-java-agent:instrumentation:servlet-2'
include ':dd-java-agent:instrumentation:servlet-3'
include ':dd-java-agent:instrumentation:sparkjava-2.4'
include ':dd-java-agent:instrumentation:spring-web'
+include ':dd-java-agent:instrumentation:spring-webflux'
include ':dd-java-agent:instrumentation:spymemcached-2.12'
include ':dd-java-agent:instrumentation:trace-annotation'

Spring webflux instrumentation

instruments spring webflux 5+