Fix Akka 10.1 support
MODIFY
dd-java-agent/instrumentation/akka-http-10.0/akka-http-10.0.gradle
dd-java-agent/instrumentation/akka-http-10.0/akka-http-10.0.gradle
@@ -9,6 +9,14 @@ apply from: "${rootDir}/gradle/test-with-scala.gradle"
apply plugin: 'org.unbroken-dome.test-sets'
testSets {
lagomTest
+
+ version101Test {
+ dirName = 'test'
+ }
+
+ latestDepTest {
+ dirName = 'test'
+ }
}
compileLagomTestJava {
@@ -26,17 +34,31 @@ muzzle {
pass {
group = 'com.typesafe.akka'
module = 'akka-http_2.11'
- versions = "[10.0.0,)"
+ versions = "[10.0.0,10.1.0)"
// later versions of akka-http expect streams to be provided
extraDependency 'com.typesafe.akka:akka-stream_2.11:2.4.14'
}
pass {
group = 'com.typesafe.akka'
module = 'akka-http_2.12'
- versions = "[10.0.0,)"
+ versions = "[10.0.0,10.1.0)"
// later versions of akka-http expect streams to be provided
extraDependency 'com.typesafe.akka:akka-stream_2.12:2.4.14'
}
+ pass {
+ group = 'com.typesafe.akka'
+ module = 'akka-http_2.11'
+ versions = "[10.1.0,)"
+ // later versions of akka-http expect streams to be provided
+ extraDependency 'com.typesafe.akka:akka-stream_2.11:2.5.11'
+ }
+ pass {
+ group = 'com.typesafe.akka'
+ module = 'akka-http_2.12'
+ versions = "[10.1.0,)"
+ // later versions of akka-http expect streams to be provided
+ extraDependency 'com.typesafe.akka:akka-stream_2.12:2.5.11'
+ }
}
dependencies {
@@ -60,6 +82,30 @@ dependencies {
lagomTestCompile project(':dd-java-agent:instrumentation:java-concurrent')
lagomTestCompile group: 'com.lightbend.lagom', name: 'lagom-javadsl-testkit_2.11', version: '1.4.0'
+
+ // There are some internal API changes in 10.1 that we would like to test separately for
+ version101TestCompile group: 'com.typesafe.akka', name: 'akka-http_2.11', version: '10.1.0'
+ version101TestCompile group: 'com.typesafe.akka', name: 'akka-stream_2.11', version: '2.5.11'
+ version101TestCompile project(':dd-java-agent:testing')
+ version101TestCompile project(':dd-java-agent:instrumentation:java-concurrent')
+ version101TestCompile project(':dd-java-agent:instrumentation:trace-annotation')
+
+ latestDepTestCompile group: 'com.typesafe.akka', name: 'akka-http_2.11', version: '+'
+ latestDepTestCompile group: 'com.typesafe.akka', name: 'akka-stream_2.11', version: '+'
+ latestDepTestCompile project(':dd-java-agent:testing')
+ latestDepTestCompile project(':dd-java-agent:instrumentation:java-concurrent')
+ latestDepTestCompile project(':dd-java-agent:instrumentation:trace-annotation')
}
test.dependsOn lagomTest
+test.dependsOn version101Test
+
+compileVersion101TestGroovy {
+ classpath = classpath.plus(files(compileVersion101TestScala.destinationDir))
+ dependsOn compileVersion101TestScala
+}
+
+compileLatestDepTestGroovy {
+ classpath = classpath.plus(files(compileLatestDepTestScala.destinationDir))
+ dependsOn compileLatestDepTestScala
+}

MODIFY
dd-java-agent/instrumentation/akka-http-10.0/src/main/scala/datadog/trace/instrumentation/akkahttp/AkkaHttpClientInstrumentation.java
dd-java-agent/instrumentation/akka-http-10.0/src/main/scala/datadog/trace/instrumentation/akkahttp/AkkaHttpClientInstrumentation.java
@@ -15,6 +15,7 @@ import com.google.auto.service.AutoService;
import datadog.trace.agent.tooling.Instrumenter;
import datadog.trace.api.DDSpanTypes;
import datadog.trace.api.DDTags;
+import datadog.trace.bootstrap.CallDepthThreadLocalMap;
import io.opentracing.Scope;
import io.opentracing.Span;
import io.opentracing.Tracer;
@@ -64,12 +65,23 @@ public final class AkkaHttpClientInstrumentation extends Instrumenter.Default {
@Override
public Map<ElementMatcher, String> transformers() {
final Map<ElementMatcher, String> transformers = new HashMap<>();
+ // This is mainly for compatibility with 10.0
transformers.put(
named("singleRequest").and(takesArgument(0, named("akka.http.scaladsl.model.HttpRequest"))),
SingleRequestAdvice.class.getName());
+ // This is for 10.1+
+ transformers.put(
+ named("singleRequestImpl")
+ .and(takesArgument(0, named("akka.http.scaladsl.model.HttpRequest"))),
+ SingleRequestAdvice.class.getName());
+ // This is mainly for compatibility with 10.0
transformers.put(
named("superPool").and(returns(named("akka.stream.scaladsl.Flow"))),
SuperPoolAdvice.class.getName());
+ // This is for 10.1+
+ transformers.put(
+ named("superPoolImpl").and(returns(named("akka.stream.scaladsl.Flow"))),
+ SuperPoolAdvice.class.getName());
return transformers;
}
@@ -77,6 +89,17 @@ public final class AkkaHttpClientInstrumentation extends Instrumenter.Default {
@Advice.OnMethodEnter(suppress = Throwable.class)
public static Scope methodEnter(
@Advice.Argument(value = 0, readOnly = false) HttpRequest request) {
+ /*
+ Versions 10.0 and 10.1 have slightly different structure that is hard to distinguish so here
+ we cast 'wider net' and avoid instrumenting twice.
+ In the future we may want to separate these, but since lots of code is reused we would need to come up
+ with way of continuing to reusing it.
+ */
+ final int callDepth = CallDepthThreadLocalMap.incrementCallDepth(HttpExt.class);
+ if (callDepth > 0) {
+ return null;
+ }
+
Tracer.SpanBuilder builder =
GlobalTracer.get()
.buildSpan("akka-http.request")
@@ -108,7 +131,13 @@ public final class AkkaHttpClientInstrumentation extends Instrumenter.Default {
@Advice.Return final Future<HttpResponse> responseFuture,
@Advice.Enter final Scope scope,
@Advice.Thrown final Throwable throwable) {
+ if (scope == null) {
+ return;
+ }
+ CallDepthThreadLocalMap.reset(HttpExt.class);
+
final Span span = scope.span();
+
if (throwable == null) {
responseFuture.onComplete(new OnCompleteHandler(span), thiz.system().dispatcher());
} else {
@@ -121,10 +150,29 @@ public final class AkkaHttpClientInstrumentation extends Instrumenter.Default {
}
public static class SuperPoolAdvice {
+
+ @Advice.OnMethodEnter(suppress = Throwable.class)
+ public static boolean methodEnter() {
+ /*
+ Versions 10.0 and 10.1 have slightly different structure that is hard to distinguish so here
+ we cast 'wider net' and avoid instrumenting twice.
+ In the future we may want to separate these, but since lots of code is reused we would need to come up
+ with way of continuing to reusing it.
+ */
+ final int callDepth = CallDepthThreadLocalMap.incrementCallDepth(HttpExt.class);
+ return callDepth <= 0;
+ }
+
@Advice.OnMethodExit(suppress = Throwable.class)
public static <T> void methodExit(
@Advice.Return(readOnly = false)
- Flow<Tuple2<HttpRequest, T>, Tuple2<Try<HttpResponse>, T>, NotUsed> flow) {
+ Flow<Tuple2<HttpRequest, T>, Tuple2<Try<HttpResponse>, T>, NotUsed> flow,
+ @Advice.Enter final boolean isApplied) {
+ if (!isApplied) {
+ return;
+ }
+ CallDepthThreadLocalMap.reset(HttpExt.class);
+
flow = AkkaHttpClientTransformFlow.transform(flow);
}
}

ADD
dd-java-agent/instrumentation/akka-http-10.0/src/test/resources/application.conf
+akka.http {
+ host-connection-pool {
+ // Limit maximum http backoff for tests
+ max-connection-backoff = 100ms
+ }
+}

Fix Akka 10.1 support