Skip to content

Commit ee3788b

Browse files
mar-kolyamar-kolya
authored
Merge pull request DataDog#672 from DataDog/mar-kolya/more-forkjoin-instrumentations
More ForkJoin instrumentations
2 parents d902ae9 + f8aed7a commit ee3788b

23 files changed

Lines changed: 1106 additions & 221 deletions

‎dd-java-agent/instrumentation/java-concurrent/akka-2.5-testing/akka-2.5-testing.gradle‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
// Set properties before any plugins get loaded
2+
ext {
3+
minJavaVersionForTests = JavaVersion.VERSION_1_8
4+
// Execute tests on all JVMs, even rare and outdated ones
5+
coreJavaInstrumentation = true
6+
}
7+
8+
apply from: "${rootDir}/gradle/java.gradle"
9+
apply from: "${rootDir}/gradle/test-with-scala.gradle"
10+
11+
dependencies {
12+
testCompile project(':dd-trace-api')
13+
testCompile project(':dd-trace-ot')
14+
testCompile deps.scala
15+
testCompile group: 'com.typesafe.akka', name: 'akka-actor_2.11', version: '2.5.0'
16+
17+
testCompile project(':dd-java-agent:testing')
18+
testCompile project(':dd-java-agent:instrumentation:java-concurrent')
19+
testCompile project(':dd-java-agent:instrumentation:trace-annotation')
20+
}

‎dd-java-agent/instrumentation/java-concurrent/akka-2.5-testing/src/test/groovy/AkkaExecutorInstrumentationTest.groovy‎

Lines changed: 149 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,149 @@
1+
import akka.dispatch.forkjoin.ForkJoinPool
2+
import akka.dispatch.forkjoin.ForkJoinTask
3+
import datadog.opentracing.DDSpan
4+
import datadog.opentracing.scopemanager.ContinuableScope
5+
import datadog.trace.agent.test.AgentTestRunner
6+
import datadog.trace.api.Trace
7+
import io.opentracing.util.GlobalTracer
8+
import spock.lang.Shared
9+
10+
import java.lang.reflect.InvocationTargetException
11+
import java.lang.reflect.Method
12+
import java.util.concurrent.ArrayBlockingQueue
13+
import java.util.concurrent.Callable
14+
import java.util.concurrent.Executor
15+
import java.util.concurrent.ExecutorService
16+
import java.util.concurrent.Future
17+
import java.util.concurrent.RejectedExecutionException
18+
import java.util.concurrent.ThreadPoolExecutor
19+
import java.util.concurrent.TimeUnit
20+
21+
/**
22+
* Test executor instrumentation for Akka specific classes.
23+
* This is to large extent a copy of ExecutorInstrumentationTest.
24+
*/
25+
class AkkaExecutorInstrumentationTest extends AgentTestRunner {
26+
@Shared
27+
Method executeRunnableMethod
28+
@Shared
29+
Method akkaExecuteForkJoinTaskMethod
30+
@Shared
31+
Method submitRunnableMethod
32+
@Shared
33+
Method submitCallableMethod
34+
@Shared
35+
Method akkaSubmitForkJoinTaskMethod
36+
@Shared
37+
Method akkaInvokeForkJoinTaskMethod
38+
39+
def setupSpec() {
40+
executeRunnableMethod = Executor.getMethod("execute", Runnable)
41+
akkaExecuteForkJoinTaskMethod = ForkJoinPool.getMethod("execute", ForkJoinTask)
42+
submitRunnableMethod = ExecutorService.getMethod("submit", Runnable)
43+
submitCallableMethod = ExecutorService.getMethod("submit", Callable)
44+
akkaSubmitForkJoinTaskMethod = ForkJoinPool.getMethod("submit", ForkJoinTask)
45+
akkaInvokeForkJoinTaskMethod = ForkJoinPool.getMethod("invoke", ForkJoinTask)
46+
}
47+
48+
// more useful name breaks java9 javac
49+
// def "#poolImpl.getClass().getSimpleName() #method.getName() propagates"()
50+
def "#poolImpl #method propagates"() {
51+
setup:
52+
def pool = poolImpl
53+
def m = method
54+
55+
new Runnable() {
56+
@Override
57+
@Trace(operationName = "parent")
58+
void run() {
59+
((ContinuableScope) GlobalTracer.get().scopeManager().active()).setAsyncPropagation(true)
60+
// this child will have a span
61+
m.invoke(pool, new AkkaAsyncChild())
62+
// this child won't
63+
m.invoke(pool, new AkkaAsyncChild(false, false))
64+
}
65+
}.run()
66+
67+
TEST_WRITER.waitForTraces(1)
68+
List<DDSpan> trace = TEST_WRITER.get(0)
69+
70+
expect:
71+
TEST_WRITER.size() == 1
72+
trace.size() == 2
73+
trace.get(0).operationName == "parent"
74+
trace.get(1).operationName == "asyncChild"
75+
trace.get(1).parentId == trace.get(0).spanId
76+
77+
cleanup:
78+
pool?.shutdown()
79+
80+
// Unfortunately, there's no simple way to test the cross product of methods/pools.
81+
where:
82+
poolImpl | method
83+
new ForkJoinPool() | executeRunnableMethod
84+
new ForkJoinPool() | akkaExecuteForkJoinTaskMethod
85+
new ForkJoinPool() | submitRunnableMethod
86+
new ForkJoinPool() | submitCallableMethod
87+
new ForkJoinPool() | akkaSubmitForkJoinTaskMethod
88+
new ForkJoinPool() | akkaInvokeForkJoinTaskMethod
89+
90+
new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1)) | executeRunnableMethod
91+
new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1)) | submitRunnableMethod
92+
new ThreadPoolExecutor(1, 1, 1000, TimeUnit.NANOSECONDS, new ArrayBlockingQueue<Runnable>(1)) | submitCallableMethod
93+
}
94+
95+
// more useful name breaks java9 javac
96+
// def "#poolImpl.getClass().getSimpleName() #method.getName() propagates"()
97+
def "#poolImpl reports after canceled jobs"() {
98+
setup:
99+
def pool = poolImpl
100+
def m = method
101+
List<AkkaAsyncChild> children = new ArrayList<>()
102+
List<Future> jobFutures = new ArrayList<>()
103+
104+
new Runnable() {
105+
@Override
106+
@Trace(operationName = "parent")
107+
void run() {
108+
((ContinuableScope) GlobalTracer.get().scopeManager().active()).setAsyncPropagation(true)
109+
try {
110+
for (int i = 0; i < 20; ++i) {
111+
// Our current instrumentation instrumentation does not behave very well
112+
// if we try to reuse Callable/Runnable. Namely we would be getting 'orphaned'
113+
// child traces sometimes since state can contain only one continuation - and
114+
// we do not really have a good way for attributing work to correct parent span
115+
// if we reuse Callable/Runnable.
116+
// Solution for now is to never reuse a Callable/Runnable.
117+
final AkkaAsyncChild child = new AkkaAsyncChild(true, true)
118+
children.add(child)
119+
try {
120+
Future f = m.invoke(pool, new AkkaAsyncChild())
121+
jobFutures.add(f)
122+
} catch (InvocationTargetException e) {
123+
throw e.getCause()
124+
}
125+
}
126+
} catch (RejectedExecutionException e) {
127+
}
128+
129+
for (Future f : jobFutures) {
130+
f.cancel(false)
131+
}
132+
for (AkkaAsyncChild child : children) {
133+
child.unblock()
134+
}
135+
}
136+
}.run()
137+
138+
TEST_WRITER.waitForTraces(1)
139+
140+
expect:
141+
// FIXME: we should improve this test to make sure continuations are actually closed
142+
TEST_WRITER.size() == 1
143+
144+
where:
145+
poolImpl | method
146+
new ForkJoinPool() | submitRunnableMethod
147+
new ForkJoinPool() | submitCallableMethod
148+
}
149+
}

‎dd-java-agent/instrumentation/java-concurrent/akka-2.5-testing/src/test/scala/AkkaAsyncChild.java‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
import akka.dispatch.forkjoin.ForkJoinTask;
2+
import datadog.trace.api.Trace;
3+
import java.util.concurrent.Callable;
4+
import java.util.concurrent.atomic.AtomicBoolean;
5+
6+
public class AkkaAsyncChild extends ForkJoinTask implements Runnable, Callable {
7+
private final AtomicBoolean blockThread;
8+
private final boolean doTraceableWork;
9+
10+
public AkkaAsyncChild() {
11+
this(true, false);
12+
}
13+
14+
@Override
15+
public Object getRawResult() {
16+
return null;
17+
}
18+
19+
@Override
20+
protected void setRawResult(final Object value) {}
21+
22+
@Override
23+
protected boolean exec() {
24+
runImpl();
25+
return true;
26+
}
27+
28+
public AkkaAsyncChild(final boolean doTraceableWork, final boolean blockThread) {
29+
this.doTraceableWork = doTraceableWork;
30+
this.blockThread = new AtomicBoolean(blockThread);
31+
}
32+
33+
public void unblock() {
34+
blockThread.set(false);
35+
}
36+
37+
@Override
38+
public void run() {
39+
runImpl();
40+
}
41+
42+
@Override
43+
public Object call() throws Exception {
44+
runImpl();
45+
return null;
46+
}
47+
48+
private void runImpl() {
49+
while (blockThread.get()) {
50+
// busy-wait to block thread
51+
}
52+
if (doTraceableWork) {
53+
asyncChild();
54+
}
55+
}
56+
57+
@Trace(operationName = "asyncChild")
58+
private void asyncChild() {}
59+
}

‎dd-java-agent/instrumentation/java-concurrent/akka-testing/akka-testing.gradle‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,9 @@
1+
// Set properties before any plugins get loaded
2+
ext {
3+
// Execute tests on all JVMs, even rare and outdated ones
4+
coreJavaInstrumentation = true
5+
}
6+
17
apply from: "${rootDir}/gradle/java.gradle"
28
apply from: "${rootDir}/gradle/test-with-scala.gradle"
39

‎dd-java-agent/instrumentation/java-concurrent/java-concurrent.gradle‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,11 @@ compileSlickTestGroovy {
2020
}
2121

2222
dependencies {
23+
// This is needed for Scala ForJoinTask/Pool instrumentation
24+
compileOnly deps.scala
25+
// This is needed for Akka ForJoinTask/Pool instrumentation
26+
compileOnly group: 'com.typesafe.akka', name: 'akka-actor_2.11', version: '2.5.0'
27+
2328
compile project(':dd-trace-api')
2429
compile project(':dd-java-agent:agent-tooling')
2530

‎dd-java-agent/instrumentation/java-concurrent/scala-testing/scala-testing.gradle‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,9 @@
1+
// Set properties before any plugins get loaded
2+
project.ext {
3+
// Execute tests on all JVMs, even rare and outdated ones
4+
coreJavaInstrumentation = true
5+
}
6+
17
apply from: "${rootDir}/gradle/java.gradle"
28
apply from: "${rootDir}/gradle/test-with-scala.gradle"
39

0 commit comments

Comments
 (0)