add tests for kt cancellation and channels
MODIFY
dd-java-agent/instrumentation/java-concurrent/kotlin-testing/src/test/groovy/KotlinCoroutineInstrumentationTest.groovy
dd-java-agent/instrumentation/java-concurrent/kotlin-testing/src/test/groovy/KotlinCoroutineInstrumentationTest.groovy
@@ -13,11 +13,39 @@ class KotlinCoroutineInstrumentationTest extends AgentTestRunner {
ThreadPoolDispatcherKt.newSingleThreadContext("Single-Thread"),
]
- // Test delay
- // Test channel
- // Test cooperative cancellation
- // Test parallel decomposition
- // EventLoop?
+ def "kotlin traced across channels"() {
+ setup:
+ KotlinCoroutineTests kotlinTest = new KotlinCoroutineTests(dispatcher)
+ int expectedNumberOfSpans = kotlinTest.tracedAcrossChannels()
+ TEST_WRITER.waitForTraces(1)
+ List<DDSpan> trace = TEST_WRITER.get(0)
+
+ expect:
+ trace.size() == expectedNumberOfSpans
+ trace[0].operationName == "KotlinCoroutineTests.tracedAcrossChannels"
+ findSpan(trace, "produce_2").context().getParentId() == trace[0].context().getSpanId()
+ findSpan(trace, "consume_2").context().getParentId() == trace[0].context().getSpanId()
+
+ where:
+ dispatcher << dispatchersToTest
+ }
+
+ def "kotlin cancellation prevents trace"() {
+ setup:
+ KotlinCoroutineTests kotlinTest = new KotlinCoroutineTests(dispatcher)
+ int expectedNumberOfSpans = kotlinTest.tracePreventedByCancellation()
+ TEST_WRITER.waitForTraces(1)
+ List<DDSpan> trace = TEST_WRITER.get(0)
+
+ expect:
+ trace.size() == expectedNumberOfSpans
+ trace[0].operationName == "KotlinCoroutineTests.tracePreventedByCancellation"
+ findSpan(trace, "preLaunch").context().getParentId() == trace[0].context().getSpanId()
+ findSpan(trace, "postLaunch") == null
+
+ where:
+ dispatcher << dispatchersToTest
+ }
def "kotlin propagates across nested jobs"() {
setup:
@@ -72,7 +100,7 @@ class KotlinCoroutineInstrumentationTest extends AgentTestRunner {
dispatcher << dispatchersToTest
}
- private DDSpan findSpan(List<DDSpan> trace, String opName) {
+ private static DDSpan findSpan(List<DDSpan> trace, String opName) {
for (DDSpan span : trace) {
if (span.getOperationName() == opName) {
return span

MODIFY
dd-java-agent/instrumentation/java-concurrent/kotlin-testing/src/test/kotlin/KotlinCoroutineTests.kt
dd-java-agent/instrumentation/java-concurrent/kotlin-testing/src/test/kotlin/KotlinCoroutineTests.kt
@@ -3,11 +3,53 @@ import datadog.trace.context.TraceScope
import io.opentracing.Scope
import io.opentracing.util.GlobalTracer
import kotlinx.coroutines.*
+import kotlinx.coroutines.channels.*
import kotlinx.coroutines.selects.select
import java.util.concurrent.TimeUnit
class KotlinCoroutineTests(private val dispatcher: CoroutineDispatcher) {
+ @Trace
+ fun tracedAcrossChannels(): Int = runTest {
+ val producer = produce {
+ repeat(3){
+ tracedChild("produce_$it")
+ send(it)
+ }
+ }
+
+ val actor = actor<Int> {
+ consumeEach {
+ tracedChild("consume_$it")
+ }
+ }
+
+ producer.toChannel(actor)
+ actor.close()
+
+ 7
+ }
+
+ @Trace
+ fun tracePreventedByCancellation(): Int {
+
+ kotlin.runCatching {
+ runTest {
+ tracedChild("preLaunch")
+
+ launch(start = CoroutineStart.UNDISPATCHED) {
+ throw Exception("Child Error")
+ }
+
+ yield()
+
+ tracedChild("postLaunch")
+ }
+ }
+
+ return 2
+ }
+
@Trace
fun tracedAcrossThreadsWithNested(): Int = runTest {
val goodDeferred = async { 1 }

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
@@ -150,9 +150,6 @@ public final class ExecutorInstrumentation extends Instrumenter.Default {
transformers.put(
named("execute").and(takesArgument(0, Runnable.class)),
SetExecuteRunnableStateAdvice.class.getName());
- transformers.put( // kotlinx.coroutines.scheduling.CoroutineScheduler
- named("dispatch").and(takesArgument(0, Runnable.class)),
- SetExecuteRunnableStateAdvice.class.getName());
transformers.put(
named("submit").and(takesArgument(0, Runnable.class)),
SetSubmitRunnableStateAdvice.class.getName());
@@ -162,6 +159,12 @@ public final class ExecutorInstrumentation extends Instrumenter.Default {
transformers.put(
nameMatches("invoke(Any|All)$").and(takesArgument(0, Callable.class)),
SetCallableStateForCallableCollectionAdvice.class.getName());
+ transformers.put( // kotlinx.coroutines.scheduling.CoroutineScheduler
+ named("dispatch")
+ .and(takesArgument(0, Runnable.class))
+ .and(takesArgument(1, named("kotlinx.coroutines.scheduling.TaskContext"))),
+ SetExecuteRunnableStateAdvice.class.getName());
+
return transformers;
}

add tests for kt cancellation and channels