Akka actor instrumentation
ADD
dd-java-agent/instrumentation/java-concurrent/akka-testing/akka-testing.gradle
+apply from: "${rootDir}/gradle/java.gradle"
+apply plugin: 'scala'
+
+dependencies {
+ compile project(':dd-trace-api')
+ compile project(':dd-trace-ot')
+ compile group: 'org.scala-lang', name: 'scala-library', version: '2.11.12'
+ compile group: 'com.typesafe.akka', name: 'akka-actor_2.11', version: '2.3.16'
+ compile group: 'com.typesafe.akka', name: 'akka-testkit_2.11', version: '2.3.16'
+
+ testCompile project(':dd-java-agent:testing')
+ testCompile project(':dd-java-agent:instrumentation:java-concurrent')
+ testCompile project(':dd-java-agent:instrumentation:trace-annotation')
+}

ADD
dd-java-agent/instrumentation/java-concurrent/akka-testing/src/main/scala/AkkaActors.scala
+import datadog.trace.api.Trace
+import akka.pattern.ask
+import io.opentracing.util.GlobalTracer
+
+import scala.concurrent.duration._
+import akka.actor.{Actor, ActorLogging, ActorRef, ActorSystem, Props}
+import akka.routing.Broadcast
+import akka.util.Timeout
+
+// ! == send-message
+object AkkaActors {
+ val system: ActorSystem = ActorSystem("helloAkka")
+
+ val printer: ActorRef = system.actorOf(Receiver.props, "receiverActor")
+
+ val howdyGreeter: ActorRef =
+ system.actorOf(Greeter.props("Howdy", printer), "howdyGreeter")
+
+ val forwarder: ActorRef = system.actorOf(Forwarder.props(printer), "forwarderActor")
+ val helloGreeter: ActorRef = system.actorOf(Greeter.props("Hello", forwarder), "helloGreeter")
+
+ @Trace
+ def tracedChild(opName: String): Unit = {
+ GlobalTracer.get().activeSpan().setOperationName(opName)
+ }
+}
+
+class AkkaActors {
+ import AkkaActors._
+ import Greeter._
+ implicit val timeout: Timeout = 5.minutes
+
+ @Trace
+ def basicTell() : Unit = {
+ howdyGreeter ! WhoToGreet("Akka")
+ howdyGreeter ! Greet
+ }
+
+ @Trace
+ def basicAsk() : Unit = {
+ howdyGreeter ! WhoToGreet("Akka")
+ howdyGreeter ? Greet
+ }
+
+ @Trace
+ def basicForward() : Unit = {
+ helloGreeter ! WhoToGreet("Akka")
+ helloGreeter ? Greet
+ }
+}
+
+object Greeter {
+ def props(message: String, receiverActor: ActorRef): Props = Props(new Greeter(message, receiverActor))
+ final case class WhoToGreet(who: String)
+ case object Greet
+}
+
+class Greeter(message: String, receiverActor: ActorRef) extends Actor {
+ import Greeter._
+ import Receiver._
+
+ var greeting = ""
+
+ def receive = {
+ case WhoToGreet(who) =>
+ greeting = s"$message, $who"
+ case Greet =>
+ receiverActor ! Greeting(greeting)
+ }
+}
+
+object Receiver {
+ def props: Props = Props[Receiver]
+ final case class Greeting(greeting: String)
+}
+
+class Receiver extends Actor with ActorLogging {
+ import Receiver._
+
+ def receive = {
+ case Greeting(greeting) => {
+ AkkaActors.tracedChild(greeting)
+ "done"
+ }
+
+ }
+}
+
+object Forwarder {
+ def props(receiverActor: ActorRef): Props = Props(new Forwarder(receiverActor))
+}
+
+class Forwarder(receiverActor: ActorRef) extends Actor with ActorLogging {
+ def receive = {
+ case msg => {
+ receiverActor forward msg
+ }
+ }
+}

ADD
dd-java-agent/instrumentation/java-concurrent/akka-testing/src/test/groovy/AkkaActorTest.groovy
+import datadog.opentracing.DDSpan
+import datadog.trace.agent.test.AgentTestRunner
+import spock.lang.Unroll
+
+class AkkaActorTest extends AgentTestRunner {
+ static {
+ System.setProperty("dd.integration.java_concurrent.enabled", "true")
+ }
+
+ @Override
+ void afterTest() {
+ // Ignore failures to instrument sun proxy classes
+ }
+
+ @Unroll
+ def "akka #testMethod" () {
+ setup:
+ AkkaActors akkaTester = new AkkaActors()
+ akkaTester."$testMethod"()
+
+ TEST_WRITER.waitForTraces(1)
+ List<DDSpan> trace = TEST_WRITER.get(0)
+
+ expect:
+ TEST_WRITER.size() == 1
+ trace.size() == 2
+ trace[0].getOperationName() == "AkkaActors.$testMethod"
+ findSpan(trace, "$expectedGreeting, Akka").context().getParentId() == trace[0].getSpanId()
+
+ where:
+ testMethod | expectedGreeting
+ "basicTell" | "Howdy"
+ "basicAsk" | "Howdy"
+ "basicForward" | "Hello"
+ }
+
+ private DDSpan findSpan(List<DDSpan> trace, String opName) {
+ for (DDSpan span : trace) {
+ if (span.getOperationName() == opName) {
+ return span
+ }
+ }
+ return null
+ }
+}

MODIFY
dd-java-agent/instrumentation/java-concurrent/scala-testing/scala-testing.gradle
dd-java-agent/instrumentation/java-concurrent/scala-testing/scala-testing.gradle
@@ -2,13 +2,9 @@ apply from: "${rootDir}/gradle/java.gradle"
apply plugin: 'scala'
dependencies {
- // calling scala classes in spock requires an explicit dependency,
- // hence the compile instead of testCompile
- compileOnly group: 'org.scala-lang', name: 'scala-library', version: '2.11.12'
- testCompile group: 'org.scala-lang', name: 'scala-library', version: '2.11.12'
-
compile project(':dd-trace-api')
compile project(':dd-trace-ot')
+ compile group: 'org.scala-lang', name: 'scala-library', version: '2.11.12'
testCompile project(':dd-java-agent:testing')
testCompile project(':dd-java-agent:instrumentation:java-concurrent')

MODIFY
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/ExecutorInstrumentation.java
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/ExecutorInstrumentation.java
@@ -139,7 +139,7 @@ public final class ExecutorInstrumentation extends Instrumenter.Configurable {
final Scope scope = GlobalTracer.get().scopeManager().active();
if (scope instanceof TraceScope && task != null && !(task instanceof DatadogWrapper)) {
task = new RunnableWrapper(task, (TraceScope) scope);
- return (DatadogWrapper) task;
+ return (RunnableWrapper) task;
}
return null;
}
@@ -160,7 +160,7 @@ public final class ExecutorInstrumentation extends Instrumenter.Configurable {
final Scope scope = GlobalTracer.get().scopeManager().active();
if (scope instanceof TraceScope && task != null && !(task instanceof DatadogWrapper)) {
task = new CallableWrapper(task, (TraceScope) scope);
- return (DatadogWrapper) task;
+ return (CallableWrapper) task;
}
return null;
}

MODIFY
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/FutureInstrumentation.java
dd-java-agent/instrumentation/java-concurrent/src/main/java/datadog/trace/instrumentation/java/concurrent/FutureInstrumentation.java
@@ -48,7 +48,6 @@ public final class FutureInstrumentation extends Instrumenter.Configurable {
"scala.concurrent.forkjoin.ForkJoinTask$AdaptedCallable",
"scala.concurrent.forkjoin.ForkJoinTask$AdaptedRunnable",
"scala.collection.parallel.AdaptiveWorkStealingForkJoinTasks$WrappedTask",
- /*
"akka.dispatch.Mailbox",
"akka.dispatch.forkjoin.ForkJoinTask",
"akka.dispatch.Mailboxes$$anon$1",
@@ -57,7 +56,6 @@ public final class FutureInstrumentation extends Instrumenter.Configurable {
"akka.dispatch.forkjoin.ForkJoinTask$AdaptedRunnable",
"akka.dispatch.Dispatcher$$anon$1",
"akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask",
- */
"com.google.common.util.concurrent.SettableFuture",
"com.google.common.util.concurrent.AbstractFuture$TrustedFuture",
"com.google.common.util.concurrent.AbstractFuture"

MODIFY
dd-java-agent/testing/src/main/java/datadog/trace/agent/test/AgentTestRunner.java
dd-java-agent/testing/src/main/java/datadog/trace/agent/test/AgentTestRunner.java
@@ -146,6 +146,8 @@ public abstract class AgentTestRunner extends Specification {
final JavaModule module,
final boolean loaded,
final Throwable throwable) {
+ // System.err.println("failed to instrument " + typeName);
+ // throwable.printStackTrace();
INSTRUMENTATION_ERROR_COUNT.incrementAndGet();
}

MODIFY
settings.gradle
settings.gradle
@@ -17,6 +17,7 @@ include ':dd-java-agent:instrumentation:datastax-cassandra-3.2'
include ':dd-java-agent:instrumentation:jax-rs'
include ':dd-java-agent:instrumentation:java-concurrent'
include ':dd-java-agent:instrumentation:java-concurrent:scala-testing'
+include ':dd-java-agent:instrumentation:java-concurrent:akka-testing'
include ':dd-java-agent:instrumentation:jboss-classloading'
include ':dd-java-agent:instrumentation:jdbc'
include ':dd-java-agent:instrumentation:jedis-1.4'

Akka actor instrumentation