diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/build.gradle b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/build.gradle index 50d136149d4..d3f2e174f5f 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/build.gradle +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/build.gradle @@ -1,18 +1,10 @@ + muzzle { pass { group = "io.reactivex.rxjava3" module = "rxjava" versions = "[3.0.0,)" } - // Assert the rxjava3 advice never resolves against rxjava2 — the two namespaces - // must not overlap. rxjava3 references io.reactivex.rxjava3.core.*, absent from the - // rxjava2 artifact, so muzzle must fail to match it. - fail { - name = "rxjava2-must-not-match" - group = "io.reactivex.rxjava2" - module = "rxjava" - versions = "[2.0.0,)" - } } apply from: "$rootDir/gradle/java.gradle" @@ -20,17 +12,13 @@ apply from: "$rootDir/gradle/java.gradle" addTestSuiteForDir('latestDepTest', 'test') dependencies { - compileOnly group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.3' + compileOnly group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.0' compileOnly group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0' testImplementation project(':dd-java-agent:instrumentation:datadog:tracing:trace-annotation') testImplementation project(':dd-java-agent:instrumentation:opentelemetry:opentelemetry-annotations-1.20') + testImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0' testImplementation group: 'io.opentelemetry.instrumentation', name: 'opentelemetry-instrumentation-annotations', version: '1.28.0' - - // Load the rxjava2 instrumenter at test runtime to prove the two versions coexist on - // the agent without interference (it stays dormant with only rxjava3 on the classpath). - testRuntimeOnly project(':dd-java-agent:instrumentation:rxjava:rxjava-2.0') - - latestDepTestImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '+' + latestDepTestImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.+' } diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/gradle.lockfile b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/gradle.lockfile deleted file mode 100644 index ae87d8ef572..00000000000 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/gradle.lockfile +++ /dev/null @@ -1,134 +0,0 @@ -# This is a Gradle generated file for dependency locking. -# Manual edits can break the build and are not advised. -# This file is expected to be part of source control. -# To regenerate this file, run: ./gradlew :dd-java-agent:instrumentation:rxjava:rxjava-3.0:dependencies --write-locks -cafe.cryptography:curve25519-elisabeth:0.1.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -cafe.cryptography:ed25519-elisabeth:0.1.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -ch.qos.logback:logback-classic:1.2.13=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -ch.qos.logback:logback-core:1.2.13=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.blogspot.mydailyjava:weak-lock-free:0.17=buildTimeInstrumentationPlugin,compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,muzzleTooling,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -com.datadoghq.okhttp3:okhttp:3.12.15=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.datadoghq.okio:okio:1.17.6=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.datadoghq:dd-instrument-java:0.0.4=buildTimeInstrumentationPlugin,compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,muzzleBootstrap,muzzleTooling,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -com.datadoghq:dd-javac-plugin-client:0.2.2=buildTimeInstrumentationPlugin,compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,muzzleBootstrap,muzzleTooling,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -com.datadoghq:java-dogstatsd-client:4.4.5=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.datadoghq:sketches-java:0.8.3=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.javaparser:javaparser-core:3.25.6=codenarc -com.github.jnr:jffi:1.3.15=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-a64asm:1.0.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-constants:0.10.4=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-enxio:0.32.20=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-ffi:2.2.19=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-posix:3.1.22=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-unixsocket:0.38.25=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.jnr:jnr-x86asm:1.0.2=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.github.spotbugs:spotbugs-annotations:4.9.8=compileClasspath,spotbugs -com.github.spotbugs:spotbugs:4.9.8=spotbugs -com.github.stephenc.jcip:jcip-annotations:1.0-1=spotbugs -com.google.auto.service:auto-service-annotations:1.1.1=annotationProcessor,compileClasspath,latestDepTestAnnotationProcessor,latestDepTestCompileClasspath,testAnnotationProcessor,testCompileClasspath -com.google.auto.service:auto-service:1.1.1=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -com.google.auto:auto-common:1.2.1=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -com.google.code.findbugs:jsr305:3.0.2=annotationProcessor,compileClasspath,latestDepTestAnnotationProcessor,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,spotbugs,testAnnotationProcessor,testCompileClasspath,testRuntimeClasspath -com.google.code.gson:gson:2.13.2=spotbugs -com.google.errorprone:error_prone_annotations:2.18.0=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -com.google.errorprone:error_prone_annotations:2.41.0=spotbugs -com.google.errorprone:error_prone_annotations:2.47.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.google.guava:failureaccess:1.0.1=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -com.google.guava:failureaccess:1.0.3=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.google.guava:guava:32.0.1-jre=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -com.google.guava:guava:33.6.0-jre=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.google.guava:listenablefuture:9999.0-empty-to-avoid-conflict-with-guava=annotationProcessor,latestDepTestAnnotationProcessor,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testAnnotationProcessor,testCompileClasspath,testRuntimeClasspath -com.google.j2objc:j2objc-annotations:2.8=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -com.google.j2objc:j2objc-annotations:3.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.google.re2j:re2j:1.8=latestDepTestRuntimeClasspath,testRuntimeClasspath -com.squareup.moshi:moshi:1.11.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.squareup.okhttp3:logging-interceptor:3.12.12=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.squareup.okhttp3:okhttp:3.12.12=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.squareup.okio:okio:1.17.5=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -com.thoughtworks.qdox:qdox:1.12.1=codenarc -commons-fileupload:commons-fileupload:1.5=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -commons-io:commons-io:2.11.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -commons-io:commons-io:2.20.0=spotbugs -de.thetaphi:forbiddenapis:3.10=compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.leangen.geantyref:geantyref:1.3.16=latestDepTestRuntimeClasspath,testRuntimeClasspath -io.opentelemetry.instrumentation:opentelemetry-instrumentation-annotations:1.28.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.opentelemetry:opentelemetry-api:1.28.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.opentelemetry:opentelemetry-context:1.28.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.reactivex.rxjava3:rxjava:3.0.0=compileClasspath,testCompileClasspath,testRuntimeClasspath -io.reactivex.rxjava3:rxjava:3.1.12=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath -io.sqreen:libsqreen:17.3.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -javax.servlet:javax.servlet-api:3.1.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -jaxen:jaxen:2.0.0=spotbugs -junit:junit:4.13.2=latestDepTestRuntimeClasspath,testRuntimeClasspath -net.bytebuddy:byte-buddy-agent:1.18.10=buildTimeInstrumentationPlugin,compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,muzzleTooling,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -net.bytebuddy:byte-buddy:1.18.10=buildTimeInstrumentationPlugin,compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,muzzleTooling,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -net.java.dev.jna:jna-platform:5.8.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -net.java.dev.jna:jna:5.8.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -net.sf.saxon:Saxon-HE:12.9=spotbugs -org.apache.ant:ant-antlr:1.10.14=codenarc -org.apache.ant:ant-junit:1.10.14=codenarc -org.apache.bcel:bcel:6.11.0=spotbugs -org.apache.commons:commons-lang3:3.19.0=spotbugs -org.apache.commons:commons-text:1.14.0=spotbugs -org.apache.logging.log4j:log4j-api:2.25.2=spotbugs -org.apache.logging.log4j:log4j-core:2.25.2=spotbugs -org.apiguardian:apiguardian-api:1.1.2=latestDepTestCompileClasspath,testCompileClasspath -org.checkerframework:checker-qual:3.33.0=annotationProcessor,latestDepTestAnnotationProcessor,testAnnotationProcessor -org.codehaus.groovy:groovy-ant:3.0.23=codenarc -org.codehaus.groovy:groovy-docgenerator:3.0.23=codenarc -org.codehaus.groovy:groovy-groovydoc:3.0.23=codenarc -org.codehaus.groovy:groovy-json:3.0.23=codenarc -org.codehaus.groovy:groovy-json:3.0.25=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.codehaus.groovy:groovy-templates:3.0.23=codenarc -org.codehaus.groovy:groovy-xml:3.0.23=codenarc -org.codehaus.groovy:groovy:3.0.23=codenarc -org.codehaus.groovy:groovy:3.0.25=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.codenarc:CodeNarc:3.7.0=codenarc -org.dom4j:dom4j:2.2.0=spotbugs -org.gmetrics:GMetrics:2.1.0=codenarc -org.hamcrest:hamcrest-core:1.3=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.hamcrest:hamcrest:3.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.jctools:jctools-core-jdk11:4.0.6=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.jctools:jctools-core:4.0.6=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.jspecify:jspecify:1.0.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.junit.jupiter:junit-jupiter-api:5.14.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.junit.jupiter:junit-jupiter-engine:5.14.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.junit.jupiter:junit-jupiter-params:5.14.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.junit.jupiter:junit-jupiter:5.14.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-commons:1.14.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-engine:1.14.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-launcher:1.14.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-runner:1.14.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-suite-api:1.14.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.junit.platform:junit-platform-suite-commons:1.14.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.junit:junit-bom:5.14.0=spotbugs -org.junit:junit-bom:5.14.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.mockito:mockito-core:4.4.0=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.objenesis:objenesis:3.3=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.opentest4j:opentest4j:1.3.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.ow2.asm:asm-analysis:9.7.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.ow2.asm:asm-analysis:9.9=spotbugs -org.ow2.asm:asm-commons:9.9=spotbugs -org.ow2.asm:asm-commons:9.9.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.ow2.asm:asm-tree:9.9=spotbugs -org.ow2.asm:asm-tree:9.9.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.ow2.asm:asm-util:9.7.1=latestDepTestRuntimeClasspath,testRuntimeClasspath -org.ow2.asm:asm-util:9.9=spotbugs -org.ow2.asm:asm:9.9=spotbugs -org.ow2.asm:asm:9.9.1=buildTimeInstrumentationPlugin,compileClasspath,latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,muzzleTooling,runtimeClasspath,testCompileClasspath,testRuntimeClasspath -org.reactivestreams:reactive-streams:1.0.3=compileClasspath,testCompileClasspath,testRuntimeClasspath -org.reactivestreams:reactive-streams:1.0.4=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath -org.slf4j:jcl-over-slf4j:1.7.30=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.slf4j:jul-to-slf4j:1.7.30=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.slf4j:log4j-over-slf4j:1.7.30=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.slf4j:slf4j-api:1.7.30=buildTimeInstrumentationPlugin,compileClasspath,muzzleBootstrap,muzzleTooling,runtimeClasspath -org.slf4j:slf4j-api:1.7.32=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.slf4j:slf4j-api:2.0.17=spotbugs,spotbugsSlf4j -org.slf4j:slf4j-simple:2.0.17=spotbugsSlf4j -org.snakeyaml:snakeyaml-engine:2.9=buildTimeInstrumentationPlugin,latestDepTestRuntimeClasspath,muzzleTooling,runtimeClasspath,testRuntimeClasspath -org.spockframework:spock-bom:2.4-groovy-3.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.spockframework:spock-core:2.4-groovy-3.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.tabletest:tabletest-junit:1.2.1=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.tabletest:tabletest-parser:1.2.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -org.xmlresolver:xmlresolver:5.3.3=spotbugs -empty=spotbugsPlugins diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/CompletableInstrumentation.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/CompletableInstrumentation.java index a77e46fe698..2b53cbaae50 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/CompletableInstrumentation.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/CompletableInstrumentation.java @@ -55,6 +55,7 @@ public static ContextScope onSubscribe( Context parentContext = InstrumentationContext.get(Completable.class, Context.class).get(completable); if (parentContext != null) { + // wrap the observer so spans from its events treat the captured span as their parent observer = new TracingCompletableObserver(observer, parentContext); // attach the context here in case additional observers are created during subscribe return parentContext.attach(); diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/FlowableInstrumentation.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/FlowableInstrumentation.java index 35f06daed79..0690ac02263 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/FlowableInstrumentation.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/FlowableInstrumentation.java @@ -13,8 +13,8 @@ import datadog.trace.agent.tooling.Instrumenter; import datadog.trace.bootstrap.InstrumentationContext; import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.FlowableSubscriber; import net.bytebuddy.asm.Advice; +import org.reactivestreams.Subscriber; public final class FlowableInstrumentation implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { @@ -31,7 +31,7 @@ public void methodAdvice(MethodTransformer transformer) { isMethod() .and(named("subscribe")) .and(takesArguments(1)) - .and(takesArgument(0, named("io.reactivex.rxjava3.core.FlowableSubscriber"))), + .and(takesArgument(0, named("org.reactivestreams.Subscriber"))), getClass().getName() + "$PropagateParentSpanAdvice"); } @@ -49,11 +49,12 @@ public static class PropagateParentSpanAdvice { @Advice.OnMethodEnter(suppress = Throwable.class) public static ContextScope onSubscribe( @Advice.This final Flowable flowable, - @Advice.Argument(value = 0, readOnly = false) FlowableSubscriber subscriber) { + @Advice.Argument(value = 0, readOnly = false) Subscriber subscriber) { if (subscriber != null) { Context parentContext = InstrumentationContext.get(Flowable.class, Context.class).get(flowable); if (parentContext != null) { + // wrap the subscriber so spans from its events treat the captured span as their parent subscriber = new TracingSubscriber<>(subscriber, parentContext); // attach the context here in case additional subscribers are created during subscribe return parentContext.attach(); diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/MaybeInstrumentation.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/MaybeInstrumentation.java index 0bc833922be..3c6eac14f8c 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/MaybeInstrumentation.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/MaybeInstrumentation.java @@ -52,6 +52,7 @@ public static ContextScope onSubscribe( if (observer != null) { Context parentContext = InstrumentationContext.get(Maybe.class, Context.class).get(maybe); if (parentContext != null) { + // wrap the observer so spans from its events treat the captured span as their parent observer = new TracingMaybeObserver<>(observer, parentContext); // attach the context here in case additional observers are created during subscribe return parentContext.attach(); diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/ObservableInstrumentation.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/ObservableInstrumentation.java index 304b710aa87..0c48f558b73 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/ObservableInstrumentation.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/ObservableInstrumentation.java @@ -53,6 +53,7 @@ public static ContextScope onSubscribe( Context parentContext = InstrumentationContext.get(Observable.class, Context.class).get(observable); if (parentContext != null) { + // wrap the observer so spans from its events treat the captured span as their parent observer = new TracingObserver<>(observer, parentContext); // attach the context here in case additional observers are created during subscribe return parentContext.attach(); diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/RxJavaModule.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/RxJavaModule.java index 69cfb0e833a..842bd3b703c 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/RxJavaModule.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/RxJavaModule.java @@ -30,13 +30,12 @@ public String[] helperClassNames() { @Override public Map contextStore() { - String contextClass = Context.class.getName(); final Map store = new HashMap<>(); - store.put("io.reactivex.rxjava3.core.Flowable", contextClass); - store.put("io.reactivex.rxjava3.core.Completable", contextClass); - store.put("io.reactivex.rxjava3.core.Maybe", contextClass); - store.put("io.reactivex.rxjava3.core.Observable", contextClass); - store.put("io.reactivex.rxjava3.core.Single", contextClass); + store.put("io.reactivex.rxjava3.core.Flowable", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Completable", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Maybe", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Observable", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Single", Context.class.getName()); return store; } diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/SingleInstrumentation.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/SingleInstrumentation.java index 3dfc1382da2..a3ba6ed53fb 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/SingleInstrumentation.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/SingleInstrumentation.java @@ -53,6 +53,7 @@ public static ContextScope onSubscribe( if (observer != null) { Context parentContext = InstrumentationContext.get(Single.class, Context.class).get(single); if (parentContext != null) { + // wrap the observer so spans from its events treat the captured span as their parent observer = new TracingSingleObserver<>(observer, parentContext); // attach the context here in case additional observers are created during subscribe return parentContext.attach(); diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingObserver.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingObserver.java index 50bad4e92a1..32018611cd1 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingObserver.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingObserver.java @@ -4,15 +4,13 @@ import datadog.context.ContextScope; import io.reactivex.rxjava3.core.Observer; import io.reactivex.rxjava3.disposables.Disposable; -import javax.annotation.Nonnull; /** Wrapper that makes sure spans from observer events treat the captured span as their parent. */ public final class TracingObserver implements Observer { private final Observer observer; private final Context parentContext; - public TracingObserver( - @Nonnull final Observer observer, @Nonnull final Context parentContext) { + public TracingObserver(final Observer observer, final Context parentContext) { this.observer = observer; this.parentContext = parentContext; } diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSubscriber.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSubscriber.java index 49caa0e6ecf..c25317ecb8a 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSubscriber.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSubscriber.java @@ -2,17 +2,17 @@ import datadog.context.Context; import datadog.context.ContextScope; -import io.reactivex.rxjava3.core.FlowableSubscriber; import javax.annotation.Nonnull; +import org.reactivestreams.Subscriber; import org.reactivestreams.Subscription; /** Wrapper that makes sure spans from subscriber events treat the captured span as their parent. */ -public final class TracingSubscriber implements FlowableSubscriber { - private final FlowableSubscriber subscriber; +public final class TracingSubscriber implements Subscriber { + private final Subscriber subscriber; private final Context parentContext; public TracingSubscriber( - @Nonnull final FlowableSubscriber subscriber, @Nonnull final Context parentContext) { + @Nonnull final Subscriber subscriber, @Nonnull final Context parentContext) { this.subscriber = subscriber; this.parentContext = parentContext; } diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/annotatedsample/RxJava3TracedMethods.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/annotatedsample/RxJava3TracedMethods.java deleted file mode 100644 index 163f5fb4b69..00000000000 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/annotatedsample/RxJava3TracedMethods.java +++ /dev/null @@ -1,137 +0,0 @@ -package annotatedsample; - -import static java.util.concurrent.TimeUnit.SECONDS; - -import io.opentelemetry.instrumentation.annotations.WithSpan; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; -import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.Single; -import java.util.concurrent.CountDownLatch; - -public class RxJava3TracedMethods { - @WithSpan - public static Completable traceAsyncCompletable(CountDownLatch latch) { - return Completable.fromRunnable(() -> await(latch)); - } - - @WithSpan - public static Completable traceAsyncFailingCompletable( - CountDownLatch latch, Exception exception) { - return Completable.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Maybe traceAsyncMaybe(CountDownLatch latch) { - return Maybe.fromCallable( - () -> { - await(latch); - return "hello"; - }); - } - - @WithSpan - public static Maybe traceAsyncFailingMaybe(CountDownLatch latch, Exception exception) { - return Maybe.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Single traceAsyncSingle(CountDownLatch latch) { - return Single.fromCallable( - () -> { - await(latch); - return "hello"; - }); - } - - @WithSpan - public static Single traceAsyncFailingSingle(CountDownLatch latch, Exception exception) { - return Single.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Observable traceAsyncObservable(CountDownLatch latch) { - return Observable.fromCallable( - () -> { - await(latch); - return "hello"; - }); - } - - @WithSpan - public static Observable traceAsyncFailingObservable( - CountDownLatch latch, Exception exception) { - return Observable.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Flowable traceAsyncFlowable(CountDownLatch latch) { - return Flowable.fromCallable( - () -> { - await(latch); - return "hello"; - }); - } - - @WithSpan - public static Flowable traceAsyncFailingFlowable( - CountDownLatch latch, Exception exception) { - return Flowable.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Completable traceAsyncNeverCompletable() { - return Completable.never(); - } - - @WithSpan - public static Maybe traceAsyncNeverMaybe() { - return Maybe.never(); - } - - @WithSpan - public static Single traceAsyncNeverSingle() { - return Single.never(); - } - - @WithSpan - public static Observable traceAsyncNeverObservable() { - return Observable.never(); - } - - @WithSpan - public static Flowable traceAsyncNeverFlowable() { - return Flowable.never(); - } - - private static void await(CountDownLatch latch) { - try { - if (!latch.await(5, SECONDS)) { - throw new IllegalStateException("Latch still locked"); - } - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } -} diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3SubscriptionTest.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3SubscriptionTest.java new file mode 100644 index 00000000000..d6b5c1c2027 --- /dev/null +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3SubscriptionTest.java @@ -0,0 +1,173 @@ +package datadog.trace.instrumentation.rxjava3; + +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; +import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import io.reactivex.rxjava3.core.Completable; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.core.Maybe; +import io.reactivex.rxjava3.core.Observable; +import io.reactivex.rxjava3.core.Single; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +/** + * Tests that RxJava 3 context propagation correctly bridges parent-child span relationships across + * subscribe boundaries. The instrumentation creates no spans of its own — it only ensures the trace + * context active at subscribe time is restored inside subscriber callbacks. + */ +class RxJava3SubscriptionTest extends AbstractInstrumentationTest { + + @Test + void maybeSubscribeLambdaRunsUnderParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AgentSpan parentSpan = startSpan("test", "parent"); + AgentScope parentScope = activateSpan(parentSpan); + try { + Maybe.create( + emitter -> { + emitter.onSuccess(new Connection()); + }) + .subscribe( + connection -> { + ((Connection) connection).query(); + latch.countDown(); + }); + } finally { + parentScope.close(); + parentSpan.finish(); + } + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("parent").root().tags(defaultTags()), + span().operationName("Connection.query").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void singleSubscribeLambdaRunsUnderParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AgentSpan parentSpan = startSpan("test", "parent"); + AgentScope parentScope = activateSpan(parentSpan); + try { + Single.create( + emitter -> { + emitter.onSuccess(new Connection()); + }) + .subscribe( + connection -> { + ((Connection) connection).query(); + latch.countDown(); + }); + } finally { + parentScope.close(); + parentSpan.finish(); + } + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("parent").root().tags(defaultTags()), + span().operationName("Connection.query").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void observableSubscribeLambdaRunsUnderParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AgentSpan parentSpan = startSpan("test", "parent"); + AgentScope parentScope = activateSpan(parentSpan); + try { + Observable.create( + emitter -> { + emitter.onNext(new Connection()); + emitter.onComplete(); + }) + .subscribe( + connection -> { + ((Connection) connection).query(); + latch.countDown(); + }); + } finally { + parentScope.close(); + parentSpan.finish(); + } + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("parent").root().tags(defaultTags()), + span().operationName("Connection.query").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void flowableSubscribeLambdaRunsUnderParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AgentSpan parentSpan = startSpan("test", "parent"); + AgentScope parentScope = activateSpan(parentSpan); + try { + Flowable.just(new Connection()) + .subscribe( + connection -> { + connection.query(); + latch.countDown(); + }); + } finally { + parentScope.close(); + parentSpan.finish(); + } + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("parent").root().tags(defaultTags()), + span().operationName("Connection.query").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void completableSubscribeLambdaRunsUnderParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AgentSpan parentSpan = startSpan("test", "parent"); + AgentScope parentScope = activateSpan(parentSpan); + try { + Completable.create( + emitter -> { + Connection connection = new Connection(); + connection.query(); + emitter.onComplete(); + }) + .subscribe(latch::countDown); + } finally { + parentScope.close(); + parentSpan.finish(); + } + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("parent").root().tags(defaultTags()), + span().operationName("Connection.query").childOfIndex(0).tags(defaultTags()))); + } + + static class Connection { + static int query() { + AgentSpan span = startSpan("test", "Connection.query"); + span.finish(); + return new Random().nextInt(); + } + } +} diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3Test.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3Test.java new file mode 100644 index 00000000000..1f3593b4406 --- /dev/null +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3Test.java @@ -0,0 +1,467 @@ +package datadog.trace.instrumentation.rxjava3; + +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; +import static datadog.trace.agent.test.assertions.TagsMatcher.error; +import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; +import static org.junit.jupiter.api.Assertions.assertEquals; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import io.reactivex.rxjava3.core.Flowable; +import io.reactivex.rxjava3.core.Maybe; +import io.reactivex.rxjava3.core.Observable; +import io.reactivex.rxjava3.core.Single; +import io.reactivex.rxjava3.schedulers.Schedulers; +import java.util.Arrays; +import java.util.List; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; + +/** + * Tests that RxJava 3 context propagation correctly bridges parent-child span relationships across + * reactive operator chains. The instrumentation creates no spans of its own — it ensures that spans + * created inside reactive callbacks are correctly parented to the span active at subscription time. + */ +class RxJava3Test extends AbstractInstrumentationTest { + + static final String EXCEPTION_MESSAGE = "test exception"; + + // --- Basic publisher tests: spans in map operators are children of publisher-parent --- + + @Test + void basicMaybeContextPropagation() { + int result = assemblePublisherUnderTrace(() -> Maybe.just(1).map(RxJava3Test::addOneFunc)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + @Test + void twoOperationsMaybeContextPropagation() { + int result = + assemblePublisherUnderTrace( + () -> Maybe.just(2).map(RxJava3Test::addOneFunc).map(RxJava3Test::addOneFunc)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + @Test + void delayedMaybeContextPropagation() { + int result = + assemblePublisherUnderTrace( + () -> Maybe.just(3).delay(100, TimeUnit.MILLISECONDS).map(RxJava3Test::addOneFunc)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + @Test + void basicFlowableContextPropagation() { + Object result = + assemblePublisherUnderTrace( + () -> Flowable.fromIterable(Arrays.asList(5, 6)).map(RxJava3Test::addOneFunc)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + @Test + void basicSingleContextPropagation() { + Object result = + assemblePublisherUnderTrace(() -> Single.just(1).map(RxJava3Test::addOneFunc).toMaybe()); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + @Test + void basicObservableContextPropagation() { + Object result = + assemblePublisherUnderTrace( + () -> Observable.just(1).map(RxJava3Test::addOneFunc).firstElement()); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + // --- Error tests: errors propagate through without breaking context --- + + @Test + void maybeErrorContextPropagation() { + RuntimeException expected = new RuntimeException(EXCEPTION_MESSAGE); + try { + assemblePublisherUnderTrace(() -> Maybe.error(expected)); + } catch (RuntimeException e) { + // expected + } + + assertTraces( + trace( + SORT_BY_START_TIME, + span() + .operationName("trace-parent") + .root() + .error() + .tags(defaultTags(), error(RuntimeException.class, EXCEPTION_MESSAGE)), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void flowableErrorContextPropagation() { + RuntimeException expected = new RuntimeException(EXCEPTION_MESSAGE); + try { + assemblePublisherUnderTrace(() -> Flowable.error(expected)); + } catch (RuntimeException e) { + // expected + } + + assertTraces( + trace( + SORT_BY_START_TIME, + span() + .operationName("trace-parent") + .root() + .error() + .tags(defaultTags(), error(RuntimeException.class, EXCEPTION_MESSAGE)), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void maybeStepErrorContextPropagation() { + try { + assemblePublisherUnderTrace( + () -> + Maybe.just(1) + .map(RxJava3Test::addOneFunc) + .map( + i -> { + throw new RuntimeException(EXCEPTION_MESSAGE); + })); + } catch (RuntimeException e) { + // expected + } + + assertTraces( + trace( + SORT_BY_START_TIME, + span() + .operationName("trace-parent") + .root() + .error() + .tags(defaultTags(), error(RuntimeException.class, EXCEPTION_MESSAGE)), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + // --- Cancel tests: cancellation does not break context --- + + @Test + void maybeCancelContextPropagation() { + cancelUnderTrace(() -> Maybe.just(1)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()))); + } + + @Test + void flowableCancelContextPropagation() { + cancelUnderTrace(() -> Flowable.fromIterable(Arrays.asList(5, 6))); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()))); + } + + // --- Chain tests: spans in concatenated publishers are children of publisher-parent --- + + @Test + void chainedMaybeContextPropagation() { + assemblePublisherUnderTrace( + () -> + Maybe.just(1) + .map(RxJava3Test::addOneFunc) + .map(RxJava3Test::addOneFunc) + .concatWith(Maybe.just(1).map(RxJava3Test::addOneFunc))); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + // --- Subscription time context: parent is determined at subscribe, not at assembly --- + + @Test + void maybeSpansParentedFromSubscriptionTime() { + Maybe maybe = Maybe.just(42).map(RxJava3Test::addOneFunc).map(RxJava3Test::addTwoFunc); + + AgentSpan traceParent = startSpan("test", "trace-parent"); + AgentScope traceScope = activateSpan(traceParent); + try { + maybe.blockingGet(); + } finally { + traceScope.close(); + traceParent.finish(); + } + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("addOne").childOfIndex(0).tags(defaultTags()), + span().operationName("addTwo").childOfIndex(0).tags(defaultTags()))); + } + + // --- Thread boundary test: context propagates across schedulers --- + + @Test + void threadBoundaryCrossing() { + assemblePublisherUnderTrace( + () -> Maybe.just(1).subscribeOn(Schedulers.io()).map(RxJava3Test::addOneFunc)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()))); + } + + // --- Intermediate span test: context at subscribe is correctly captured --- + + @Test + void intermediateSpanContextPropagation() { + assemblePublisherUnderTraceWithIntermediate(() -> Maybe.just(1).map(RxJava3Test::addOneFunc)); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().operationName("trace-parent").root().tags(defaultTags()), + span().operationName("publisher-parent").childOfIndex(0).tags(defaultTags()), + span().operationName("intermediate").childOfIndex(1).tags(defaultTags()), + span().operationName("addOne").childOfIndex(1).tags(defaultTags()), + span().operationName("addTwo").childOfIndex(1).tags(defaultTags()))); + } + + // --- Parallel flowable: library-behavior smoke tests --- + // These verify that parallel flowable operations complete correctly under instrumentation + // (i.e., the agent doesn't break library functionality). Context propagation for parallel + // streams is inherently non-deterministic and is covered by the sequential tests above. + + @Test + void parallelFlowableOnNewThreadScheduler() { + List values = + Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) + .parallel() + .runOn(Schedulers.newThread()) + .flatMap( + num -> + Maybe.just(num.toString() + " on " + Thread.currentThread().getName()) + .toFlowable()) + .sequential() + .toList() + .blockingGet(); + + assertEquals(4, values.size(), "Expected 4 values, got " + values.size()); + } + + @Test + void parallelFlowableOnComputationScheduler() { + List values = + Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) + .parallel() + .runOn(Schedulers.computation()) + .flatMap( + num -> + Maybe.just(num.toString() + " on " + Thread.currentThread().getName()) + .toFlowable()) + .sequential() + .toList() + .blockingGet(); + + assertEquals(4, values.size(), "Expected 4 values, got " + values.size()); + } + + // --- Helper methods --- + + @SuppressWarnings("unchecked") + private T assemblePublisherUnderTrace(PublisherSupplier publisherSupplier) { + AgentSpan traceParent = startSpan("test", "trace-parent"); + AgentScope traceScope = activateSpan(traceParent); + + AgentSpan span = startSpan("test", "publisher-parent"); + AgentScope scope = activateSpan(span); + + Object publisher = publisherSupplier.get(); + try { + if (publisher instanceof Maybe) { + return (T) ((Maybe) publisher).blockingGet(); + } else if (publisher instanceof Flowable) { + return (T) ((Flowable) publisher).toList().blockingGet(); + } else if (publisher instanceof Single) { + return (T) ((Single) publisher).blockingGet(); + } else if (publisher instanceof Observable) { + return (T) ((Observable) publisher).toList().blockingGet(); + } + throw new RuntimeException("Unknown publisher: " + publisher); + } catch (RuntimeException e) { + traceParent.setError(true); + traceParent.addThrowable(e); + throw e; + } finally { + span.finish(); + scope.close(); + traceScope.close(); + traceParent.finish(); + } + } + + private void cancelUnderTrace(PublisherSupplier publisherSupplier) { + AgentSpan traceParent = startSpan("test", "trace-parent"); + AgentScope traceScope = activateSpan(traceParent); + + AgentSpan span = startSpan("test", "publisher-parent"); + AgentScope scope = activateSpan(span); + + Object publisher = publisherSupplier.get(); + Flowable flowable; + if (publisher instanceof Maybe) { + flowable = ((Maybe) publisher).toFlowable(); + } else if (publisher instanceof Flowable) { + flowable = (Flowable) publisher; + } else { + throw new RuntimeException("Unknown publisher: " + publisher); + } + + flowable.subscribe( + new Subscriber() { + @Override + public void onSubscribe(Subscription subscription) { + subscription.cancel(); + } + + @Override + public void onNext(Object t) {} + + @Override + public void onError(Throwable error) {} + + @Override + public void onComplete() {} + }); + + scope.close(); + span.finish(); + traceScope.close(); + traceParent.finish(); + } + + @SuppressWarnings("unchecked") + private T assemblePublisherUnderTraceWithIntermediate( + PublisherSupplier publisherSupplier) { + AgentSpan traceParent = startSpan("test", "trace-parent"); + AgentScope traceScope = activateSpan(traceParent); + + AgentSpan publisherParent = startSpan("test", "publisher-parent"); + AgentScope publisherScope = activateSpan(publisherParent); + + Object publisher = publisherSupplier.get(); + + AgentSpan intermediate = startSpan("test", "intermediate"); + AgentScope intermediateScope = activateSpan(intermediate); + try { + if (publisher instanceof Maybe) { + publisher = ((Maybe) publisher).map(RxJava3Test::addTwoFunc); + } else if (publisher instanceof Flowable) { + publisher = ((Flowable) publisher).map(RxJava3Test::addTwoFunc); + } + } finally { + intermediate.finish(); + intermediateScope.close(); + } + + try { + if (publisher instanceof Maybe) { + return (T) ((Maybe) publisher).blockingGet(); + } else if (publisher instanceof Flowable) { + return (T) ((Flowable) publisher).toList().blockingGet(); + } + throw new RuntimeException("Unknown publisher: " + publisher); + } finally { + publisherParent.finish(); + publisherScope.close(); + traceScope.close(); + traceParent.finish(); + } + } + + static int addOneFunc(int i) { + AgentSpan span = startSpan("test", "addOne"); + AgentScope scope = activateSpan(span); + try { + return i + 1; + } finally { + scope.close(); + span.finish(); + } + } + + static int addTwoFunc(int i) { + AgentSpan span = startSpan("test", "addTwo"); + AgentScope scope = activateSpan(span); + try { + return i + 2; + } finally { + scope.close(); + span.finish(); + } + } + + @FunctionalInterface + interface PublisherSupplier { + Object get(); + } +} diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3InteropTest.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3InteropTest.java deleted file mode 100644 index 14ddd2f1f7b..00000000000 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3InteropTest.java +++ /dev/null @@ -1,132 +0,0 @@ -package testdog.trace.instrumentation.rxjava3; - -import static datadog.trace.agent.test.assertions.Matchers.validates; -import static datadog.trace.agent.test.assertions.SpanMatcher.span; -import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; -import static datadog.trace.agent.test.assertions.TagsMatcher.tag; -import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; -import static datadog.trace.agent.test.assertions.TraceMatcher.trace; -import static org.junit.jupiter.api.Assertions.assertEquals; - -import datadog.trace.agent.test.AbstractInstrumentationTest; -import datadog.trace.agent.test.assertions.TagsMatcher; -import datadog.trace.api.Trace; -import datadog.trace.bootstrap.instrumentation.api.Tags; -import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; -import io.reactivex.rxjava3.core.Single; -import java.util.List; -import java.util.Optional; -import java.util.concurrent.CompletableFuture; -import java.util.function.Supplier; -import java.util.stream.Stream; -import org.junit.jupiter.api.Test; - -class RxJava3InteropTest extends AbstractInstrumentationTest { - - // The component tag is stored as a UTF8BytesString, so compare by string content. - static TagsMatcher componentTrace() { - return tag(Tags.COMPONENT, validates(o -> "trace".equals(String.valueOf(o)))); - } - - static class Worker { - static int child(int i) { - return childTraced(i); - } - - @Trace(operationName = "child", resourceName = "child") - static int childTraced(int i) { - return i + 1; - } - - @Trace(operationName = "interop-parent", resourceName = "interop-parent") - static T runUnderParent(Supplier work) { - return work.get(); - } - } - - @Test - void fromCompletionStageSync() { - Integer result = - Worker.runUnderParent( - () -> - Single.fromCompletionStage(CompletableFuture.completedFuture(1)) - .map(Worker::child) - .blockingGet()); - assertEquals(2, result); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("interop-parent").resourceName("interop-parent"), - span() - .childOfIndex(0) - .operationName("child") - .resourceName("child") - .tags(componentTrace(), defaultTags()))); - } - - @Test - void fromCompletionStageAsync() { - Integer result = - Worker.runUnderParent( - () -> - Single.fromCompletionStage(CompletableFuture.supplyAsync(() -> 1)) - .map(Worker::child) - .blockingGet()); - assertEquals(2, result); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("interop-parent").resourceName("interop-parent"), - span() - .childOfIndex(0) - .operationName("child") - .resourceName("child") - .tags(componentTrace(), defaultTags()))); - } - - @Test - void fromOptional() { - Integer result = - Worker.runUnderParent( - () -> Maybe.fromOptional(Optional.of(1)).map(Worker::child).blockingGet()); - assertEquals(2, result); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("interop-parent").resourceName("interop-parent"), - span() - .childOfIndex(0) - .operationName("child") - .resourceName("child") - .tags(componentTrace(), defaultTags()))); - } - - @Test - void fromStream() { - List result = - Worker.runUnderParent( - () -> Flowable.fromStream(Stream.of(1, 2)).map(Worker::child).toList().blockingGet()); - assertEquals(2, result.size()); - assertEquals(2, result.get(0)); - assertEquals(3, result.get(1)); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("interop-parent").resourceName("interop-parent"), - span() - .childOfIndex(0) - .operationName("child") - .resourceName("child") - .tags(componentTrace(), defaultTags()), - span() - .childOfIndex(0) - .operationName("child") - .resourceName("child") - .tags(componentTrace(), defaultTags()))); - } -} diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3ResultExtensionTest.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3ResultExtensionTest.java deleted file mode 100644 index 27d76f80c1b..00000000000 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3ResultExtensionTest.java +++ /dev/null @@ -1,241 +0,0 @@ -package testdog.trace.instrumentation.rxjava3; - -import static datadog.trace.agent.test.assertions.Matchers.validates; -import static datadog.trace.agent.test.assertions.SpanMatcher.span; -import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; -import static datadog.trace.agent.test.assertions.TagsMatcher.error; -import static datadog.trace.agent.test.assertions.TagsMatcher.tag; -import static datadog.trace.agent.test.assertions.TraceMatcher.trace; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertThrows; - -import annotatedsample.RxJava3TracedMethods; -import datadog.trace.agent.test.AbstractInstrumentationTest; -import datadog.trace.agent.test.assertions.SpanMatcher; -import datadog.trace.agent.test.assertions.TagsMatcher; -import datadog.trace.bootstrap.instrumentation.api.Tags; -import datadog.trace.junit.utils.config.WithConfig; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; -import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.Single; -import java.util.concurrent.CountDownLatch; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.EnumSource; - -@WithConfig(key = "trace.otel.enabled", value = "true") -class RxJava3ResultExtensionTest extends AbstractInstrumentationTest { - - static final String EXCEPTION_MESSAGE = "Test exception"; - - // The COMPONENT and SPAN_KIND tags are stored as UTF8BytesString, so we compare by string content - // rather than using is("...") which would fail the asymmetric String#equals(UTF8BytesString) - // check. - static TagsMatcher otelComponent() { - return tag(Tags.COMPONENT, validates(o -> "opentelemetry".equals(String.valueOf(o)))); - } - - static TagsMatcher internalSpanKind() { - return tag(Tags.SPAN_KIND, validates(o -> Tags.SPAN_KIND_INTERNAL.equals(String.valueOf(o)))); - } - - // The operation and resource names are stored as UTF8BytesString, so we compare by string content - // (CharSequence equality is asymmetric: String#equals(UTF8BytesString) is false). - static SpanMatcher otelSpan(String name) { - return span() - .operationName(java.util.regex.Pattern.compile(java.util.regex.Pattern.quote(name))) - .resourceName((CharSequence cs) -> name.contentEquals(cs)); - } - - /** - * The five reactive types exercised by the test, with their type-specific terminal operations. - */ - enum ReactiveType { - COMPLETABLE("Completable"), - MAYBE("Maybe"), - SINGLE("Single"), - OBSERVABLE("Observable"), - FLOWABLE("Flowable"); - - final String type; - - ReactiveType(String type) { - this.type = type; - } - - /** Runs the blocking terminal operation that drives the async result to completion. */ - void runTerminal(Object asyncType) { - switch (this) { - case COMPLETABLE: - ((Completable) asyncType).blockingAwait(); - break; - case MAYBE: - ((Maybe) asyncType).blockingGet(); - break; - case SINGLE: - ((Single) asyncType).blockingGet(); - break; - case OBSERVABLE: - ((Observable) asyncType).blockingLast(); - break; - case FLOWABLE: - ((Flowable) asyncType).blockingLast(); - break; - default: - throw new IllegalStateException("Unknown type: " + this); - } - } - - /** Subscribes and immediately disposes (cancels) the async result. */ - void subscribeAndDispose(Object asyncType) { - switch (this) { - case COMPLETABLE: - ((Completable) asyncType).subscribe().dispose(); - break; - case MAYBE: - ((Maybe) asyncType).subscribe().dispose(); - break; - case SINGLE: - ((Single) asyncType).subscribe().dispose(); - break; - case OBSERVABLE: - ((Observable) asyncType).subscribe().dispose(); - break; - case FLOWABLE: - ((Flowable) asyncType).subscribe().dispose(); - break; - default: - throw new IllegalStateException("Unknown type: " + this); - } - } - - Object traceAsync(CountDownLatch latch) { - switch (this) { - case COMPLETABLE: - return RxJava3TracedMethods.traceAsyncCompletable(latch); - case MAYBE: - return RxJava3TracedMethods.traceAsyncMaybe(latch); - case SINGLE: - return RxJava3TracedMethods.traceAsyncSingle(latch); - case OBSERVABLE: - return RxJava3TracedMethods.traceAsyncObservable(latch); - case FLOWABLE: - return RxJava3TracedMethods.traceAsyncFlowable(latch); - default: - throw new IllegalStateException("Unknown type: " + this); - } - } - - Object traceAsyncNever() { - switch (this) { - case COMPLETABLE: - return RxJava3TracedMethods.traceAsyncNeverCompletable(); - case MAYBE: - return RxJava3TracedMethods.traceAsyncNeverMaybe(); - case SINGLE: - return RxJava3TracedMethods.traceAsyncNeverSingle(); - case OBSERVABLE: - return RxJava3TracedMethods.traceAsyncNeverObservable(); - case FLOWABLE: - return RxJava3TracedMethods.traceAsyncNeverFlowable(); - default: - throw new IllegalStateException("Unknown type: " + this); - } - } - - Object traceAsyncFailing(CountDownLatch latch, Exception exception) { - switch (this) { - case COMPLETABLE: - return RxJava3TracedMethods.traceAsyncFailingCompletable(latch, exception); - case MAYBE: - return RxJava3TracedMethods.traceAsyncFailingMaybe(latch, exception); - case SINGLE: - return RxJava3TracedMethods.traceAsyncFailingSingle(latch, exception); - case OBSERVABLE: - return RxJava3TracedMethods.traceAsyncFailingObservable(latch, exception); - case FLOWABLE: - return RxJava3TracedMethods.traceAsyncFailingFlowable(latch, exception); - default: - throw new IllegalStateException("Unknown type: " + this); - } - } - } - - @ParameterizedTest(name = "test WithSpan annotated async method {0}") - @EnumSource(ReactiveType.class) - void success(ReactiveType type) { - CountDownLatch latch = new CountDownLatch(1); - Object asyncType = type.traceAsync(latch); - - // The span must not be finished before the async result completes. - assertEquals(0, writer.size()); - - latch.countDown(); - type.runTerminal(asyncType); - - String method = "traceAsync" + type.type; - assertTraces( - trace( - otelSpan("RxJava3TracedMethods." + method) - .tags(defaultTags(), otelComponent(), internalSpanKind()))); - } - - @ParameterizedTest(name = "test WithSpan annotated async method failing {0}") - @EnumSource(ReactiveType.class) - void failing(ReactiveType type) { - CountDownLatch latch = new CountDownLatch(1); - IllegalStateException expectedException = new IllegalStateException(EXCEPTION_MESSAGE); - Object asyncType = type.traceAsyncFailing(latch, expectedException); - - assertEquals(0, writer.size()); - - latch.countDown(); - assertThrows(IllegalStateException.class, () -> type.runTerminal(asyncType)); - - String method = "traceAsyncFailing" + type.type; - assertTraces( - trace( - otelSpan("RxJava3TracedMethods." + method) - .error() - .tags( - defaultTags(), - otelComponent(), - internalSpanKind(), - error(IllegalStateException.class, EXCEPTION_MESSAGE)))); - } - - @ParameterizedTest(name = "test WithSpan annotated async method cancelled {0}") - @EnumSource(ReactiveType.class) - void cancelled(ReactiveType type) { - CountDownLatch latch = new CountDownLatch(1); - Object asyncType = type.traceAsync(latch); - - assertEquals(0, writer.size()); - - latch.countDown(); - type.subscribeAndDispose(asyncType); - - String method = "traceAsync" + type.type; - assertTraces( - trace( - otelSpan("RxJava3TracedMethods." + method) - .tags(defaultTags(), otelComponent(), internalSpanKind()))); - } - - @ParameterizedTest(name = "test WithSpan annotated never async method cancelled {0}") - @EnumSource(ReactiveType.class) - void cancelledNever(ReactiveType type) { - Object asyncType = type.traceAsyncNever(); - - assertEquals(0, writer.size()); - - type.subscribeAndDispose(asyncType); - - String method = "traceAsyncNever" + type.type; - assertTraces( - trace( - otelSpan("RxJava3TracedMethods." + method) - .tags(defaultTags(), otelComponent(), internalSpanKind()))); - } -} diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3Test.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3Test.java deleted file mode 100644 index e9bef9e2068..00000000000 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/RxJava3Test.java +++ /dev/null @@ -1,657 +0,0 @@ -package testdog.trace.instrumentation.rxjava3; - -import static datadog.trace.agent.test.assertions.Matchers.validates; -import static datadog.trace.agent.test.assertions.SpanMatcher.span; -import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; -import static datadog.trace.agent.test.assertions.TagsMatcher.error; -import static datadog.trace.agent.test.assertions.TagsMatcher.tag; -import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; -import static datadog.trace.agent.test.assertions.TraceMatcher.trace; -import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; -import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan; -import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; -import static java.util.concurrent.TimeUnit.MILLISECONDS; -import static org.junit.jupiter.api.Assertions.assertArrayEquals; -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertThrows; - -import datadog.trace.agent.test.AbstractInstrumentationTest; -import datadog.trace.agent.test.assertions.SpanMatcher; -import datadog.trace.agent.test.assertions.TagsMatcher; -import datadog.trace.api.Trace; -import datadog.trace.bootstrap.instrumentation.api.AgentScope; -import datadog.trace.bootstrap.instrumentation.api.AgentSpan; -import datadog.trace.bootstrap.instrumentation.api.Tags; -import io.reactivex.rxjava3.core.BackpressureStrategy; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; -import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.Scheduler; -import io.reactivex.rxjava3.core.Single; -import io.reactivex.rxjava3.schedulers.Schedulers; -import java.util.Arrays; -import java.util.List; -import java.util.function.Supplier; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.Arguments; -import org.junit.jupiter.params.provider.MethodSource; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; - -class RxJava3Test extends AbstractInstrumentationTest { - - static final String EXCEPTION_MESSAGE = "test exception"; - - // The component tag is stored as a UTF8BytesString, so we compare by string content rather than - // using is("trace") which would fail the asymmetric String#equals(UTF8BytesString) check. - static TagsMatcher componentTrace() { - return tag(Tags.COMPONENT, validates(o -> "trace".equals(String.valueOf(o)))); - } - - static class Worker { - static long traceParentId; - static long publisherParentId; - static long intermediateId; - - static int addOne(int i) { - return addOneTraced(i); - } - - @Trace(operationName = "addOne", resourceName = "addOne") - static int addOneTraced(int i) { - return i + 1; - } - - static int addTwo(int i) { - return addTwoTraced(i); - } - - @Trace(operationName = "addTwo", resourceName = "addTwo") - static int addTwoTraced(int i) { - return i + 2; - } - - static Object throwException() { - throw new RuntimeException(EXCEPTION_MESSAGE); - } - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - @SuppressWarnings("unchecked") - static Object assemblePublisherUnderTrace(Supplier publisherSupplier) { - traceParentId = activeSpan().getSpanId(); - AgentSpan span = startSpan("test", "publisher-parent"); - publisherParentId = span.getSpanId(); - AgentScope scope = activateSpan(span); - - Object publisher = publisherSupplier.get(); - try { - if (publisher instanceof Maybe) { - return ((Maybe) publisher).blockingGet(); - } else if (publisher instanceof Flowable) { - List list = ((Flowable) publisher).toList().blockingGet(); - return list.toArray(new Object[0]); - } - throw new RuntimeException("Unknown publisher: " + publisher); - } finally { - span.finish(); - scope.close(); - } - } - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - static void cancelUnderTrace(Supplier publisherSupplier) { - traceParentId = activeSpan().getSpanId(); - AgentSpan span = startSpan("test", "publisher-parent"); - publisherParentId = span.getSpanId(); - AgentScope scope = activateSpan(span); - - // Normalize every reactive type to a Flowable so a single Subscriber can cancel the - // subscription right away, exercising the cancellation path of each instrumentation. - Object publisher = publisherSupplier.get(); - Flowable flowable; - if (publisher instanceof Maybe) { - flowable = ((Maybe) publisher).toFlowable(); - } else if (publisher instanceof Single) { - flowable = ((Single) publisher).toFlowable(); - } else if (publisher instanceof Observable) { - flowable = ((Observable) publisher).toFlowable(BackpressureStrategy.BUFFER); - } else if (publisher instanceof Completable) { - flowable = ((Completable) publisher).toFlowable(); - } else { - flowable = (Flowable) publisher; - } - - try { - flowable.subscribe( - new Subscriber() { - @Override - public void onSubscribe(Subscription subscription) { - subscription.cancel(); - } - - @Override - public void onNext(Object t) {} - - @Override - public void onError(Throwable error) {} - - @Override - public void onComplete() {} - }); - } finally { - scope.close(); - span.finish(); - } - } - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - static Object runUnderTraceParent(Supplier work) { - traceParentId = activeSpan().getSpanId(); - return work.get(); - } - } - - // --- Publisher success --------------------------------------------------- - - static List publisherSuccessArgs() { - return Arrays.asList( - Arguments.of( - "basic maybe", - new Object[] {2}, - 1, - (Supplier) () -> Maybe.just(1).map(Worker::addOne)), - Arguments.of( - "two operations maybe", - new Object[] {4}, - 2, - (Supplier) () -> Maybe.just(2).map(Worker::addOne).map(Worker::addOne)), - Arguments.of( - "delayed maybe", - new Object[] {4}, - 1, - (Supplier) () -> Maybe.just(3).delay(100, MILLISECONDS).map(Worker::addOne)), - Arguments.of( - "delayed twice maybe", - new Object[] {6}, - 2, - (Supplier) - () -> - Maybe.just(4) - .delay(100, MILLISECONDS) - .map(Worker::addOne) - .delay(100, MILLISECONDS) - .map(Worker::addOne)), - Arguments.of( - "basic flowable", - new Object[] {6, 7}, - 2, - (Supplier) - () -> Flowable.fromIterable(Arrays.asList(5, 6)).map(Worker::addOne)), - Arguments.of( - "two operations flowable", - new Object[] {8, 9}, - 4, - (Supplier) - () -> - Flowable.fromIterable(Arrays.asList(6, 7)) - .map(Worker::addOne) - .map(Worker::addOne)), - Arguments.of( - "delayed flowable", - new Object[] {8, 9}, - 2, - (Supplier) - () -> - Flowable.fromIterable(Arrays.asList(7, 8)) - .delay(100, MILLISECONDS) - .map(Worker::addOne)), - Arguments.of( - "delayed twice flowable", - new Object[] {10, 11}, - 4, - (Supplier) - () -> - Flowable.fromIterable(Arrays.asList(8, 9)) - .delay(100, MILLISECONDS) - .map(Worker::addOne) - .delay(100, MILLISECONDS) - .map(Worker::addOne)), - Arguments.of( - "maybe from callable", - new Object[] {12}, - 2, - (Supplier) - () -> Maybe.fromCallable(() -> Worker.addOne(10)).map(Worker::addOne))); - } - - @ParameterizedTest(name = "Publisher ''{0}'' test") - @MethodSource("publisherSuccessArgs") - void publisherSuccess(String name, Object[] expected, int workSpans, Supplier supplier) { - Object result = Worker.assemblePublisherUnderTrace(supplier); - - if (expected.length == 1) { - assertEquals(expected[0], result); - } else { - assertArrayEquals(expected, (Object[]) result); - } - - SpanMatcher[] matchers = new SpanMatcher[workSpans + 2]; - matchers[0] = - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .tags(componentTrace(), defaultTags()); - matchers[1] = - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()); - for (int i = 0; i < workSpans; i++) { - matchers[2 + i] = - span() - .childOf(Worker.publisherParentId) - .operationName("addOne") - .resourceName("addOne") - .tags(componentTrace(), defaultTags()); - } - - assertTraces(trace(SORT_BY_START_TIME, matchers)); - } - - // --- Publisher error ----------------------------------------------------- - - static List publisherErrorArgs() { - return Arrays.asList( - Arguments.of( - "maybe", (Supplier) () -> Maybe.error(new RuntimeException(EXCEPTION_MESSAGE))), - Arguments.of( - "flowable", - (Supplier) () -> Flowable.error(new RuntimeException(EXCEPTION_MESSAGE)))); - } - - @ParameterizedTest(name = "Publisher error ''{0}'' test") - @MethodSource("publisherErrorArgs") - void publisherError(String name, Supplier supplier) { - RuntimeException exception = - assertThrows(RuntimeException.class, () -> Worker.assemblePublisherUnderTrace(supplier)); - assertEquals(EXCEPTION_MESSAGE, exception.getMessage()); - - assertTraces( - trace( - SORT_BY_START_TIME, - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .error() - .tags( - componentTrace(), - error(RuntimeException.class, EXCEPTION_MESSAGE), - defaultTags()), - // It's important that we don't attach errors at the reactive level so that we don't - // impact the spans on reactive integrations such as netty and lettuce. - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()))); - } - - // --- Publisher step error ------------------------------------------------ - - static List publisherStepErrorArgs() { - return Arrays.asList( - Arguments.of( - "basic maybe failure", - 1, - (Supplier) - () -> Maybe.just(1).map(Worker::addOne).map(i -> Worker.throwException())), - Arguments.of( - "basic flowable failure", - 1, - (Supplier) - () -> - Flowable.fromIterable(Arrays.asList(5, 6)) - .map(Worker::addOne) - .map(i -> Worker.throwException()))); - } - - @ParameterizedTest(name = "Publisher step ''{0}'' test") - @MethodSource("publisherStepErrorArgs") - void publisherStepError(String name, int workSpans, Supplier supplier) { - RuntimeException exception = - assertThrows(RuntimeException.class, () -> Worker.assemblePublisherUnderTrace(supplier)); - assertEquals(EXCEPTION_MESSAGE, exception.getMessage()); - - SpanMatcher[] matchers = new SpanMatcher[workSpans + 2]; - matchers[0] = - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .error() - .tags( - componentTrace(), error(RuntimeException.class, EXCEPTION_MESSAGE), defaultTags()); - matchers[1] = - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()); - for (int i = 0; i < workSpans; i++) { - matchers[2 + i] = - span() - .childOf(Worker.publisherParentId) - .operationName("addOne") - .resourceName("addOne") - .tags(componentTrace(), defaultTags()); - } - - assertTraces(trace(SORT_BY_START_TIME, matchers)); - } - - // --- Cancel -------------------------------------------------------------- - - static List cancelArgs() { - return Arrays.asList( - Arguments.of("basic maybe", (Supplier) () -> Maybe.just(1)), - Arguments.of( - "basic flowable", (Supplier) () -> Flowable.fromIterable(Arrays.asList(5, 6))), - Arguments.of("basic single", (Supplier) () -> Single.just(1)), - Arguments.of( - "basic observable", - (Supplier) () -> Observable.fromIterable(Arrays.asList(5, 6))), - Arguments.of("basic completable", (Supplier) Completable::complete)); - } - - @ParameterizedTest(name = "Publisher ''{0}'' cancel") - @MethodSource("cancelArgs") - void cancel(String name, Supplier supplier) { - Worker.cancelUnderTrace(supplier); - - assertTraces( - trace( - SORT_BY_START_TIME, - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .tags(componentTrace(), defaultTags()), - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()))); - } - - // --- Chain spans correct parent ------------------------------------------ - - static List chainParentArgs() { - return Arrays.asList( - Arguments.of( - "basic maybe", - 3, - (Supplier) - () -> - Maybe.just(1) - .map(Worker::addOne) - .map(Worker::addOne) - .concatWith(Maybe.just(1).map(Worker::addOne))), - Arguments.of( - "basic flowable", - 5, - (Supplier) - () -> - Flowable.fromIterable(Arrays.asList(5, 6)) - .map(Worker::addOne) - .map(Worker::addOne) - .concatWith(Maybe.just(1).map(Worker::addOne).toFlowable()))); - } - - @ParameterizedTest(name = "Publisher chain spans have the correct parent for ''{0}''") - @MethodSource("chainParentArgs") - void chainParent(String name, int workSpans, Supplier supplier) { - Worker.assemblePublisherUnderTrace(supplier); - - SpanMatcher[] matchers = new SpanMatcher[workSpans + 2]; - matchers[0] = - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .tags(componentTrace(), defaultTags()); - matchers[1] = - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()); - for (int i = 0; i < workSpans; i++) { - matchers[2 + i] = - span() - .childOf(Worker.publisherParentId) - .operationName("addOne") - .resourceName("addOne") - .tags(componentTrace(), defaultTags()); - } - - assertTraces(trace(SORT_BY_START_TIME, matchers)); - } - - // --- Correct parents from subscription time (blockingGet) ---------------- - - @Test - void correctParentsFromSubscriptionTimeBlockingGet() { - Maybe maybe = Maybe.just(42).map(Worker::addOne).map(Worker::addTwo); - - Worker.runUnderTraceParent( - () -> { - maybe.blockingGet(); - return null; - }); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("trace-parent").resourceName("trace-parent"), - span() - .childOf(Worker.traceParentId) - .operationName("addOne") - .tags(componentTrace(), defaultTags()), - span() - .childOf(Worker.traceParentId) - .operationName("addTwo") - .tags(componentTrace(), defaultTags()))); - } - - // --- Correct parents from subscription time (intermediate span) ---------- - - static List subscriptionTimeIntermediateArgs() { - return Arrays.asList( - Arguments.of("basic maybe", 1, (Supplier) () -> Maybe.just(1).map(Worker::addOne)), - Arguments.of( - "basic flowable", - 2, - (Supplier) - () -> Flowable.fromIterable(Arrays.asList(1, 2)).map(Worker::addOne))); - } - - @ParameterizedTest( - name = "Publisher chain spans have the correct parents from subscription time ''{0}''") - @MethodSource("subscriptionTimeIntermediateArgs") - @SuppressWarnings("unchecked") - void correctParentsFromSubscriptionTime(String name, int workItems, Supplier supplier) { - Worker.assemblePublisherUnderTrace( - () -> { - // The "add one" operations are assembled under publisher-parent and stay its children. - // The "add two" operations are assembled under intermediate, but intermediate is finished - // before subscription, so re-activating its context at delivery time is a no-op and - // addTwo falls back to the still-active publisher-parent as well. - Object publisher = supplier.get(); - - AgentSpan intermediate = startSpan("test", "intermediate"); - Worker.intermediateId = intermediate.getSpanId(); - AgentScope scope = activateSpan(intermediate); - try { - if (publisher instanceof Maybe) { - return ((Maybe) publisher).map(Worker::addTwo); - } else if (publisher instanceof Flowable) { - return ((Flowable) publisher).map(Worker::addTwo); - } - throw new IllegalStateException("Unknown publisher type"); - } finally { - intermediate.finish(); - scope.close(); - } - }); - - SpanMatcher[] matchers = new SpanMatcher[3 + 2 * workItems]; - matchers[0] = - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .tags(componentTrace(), defaultTags()); - matchers[1] = - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()); - matchers[2] = - span() - .id(Worker.intermediateId) - .childOf(Worker.publisherParentId) - .operationName("intermediate") - .resourceName("intermediate") - .tags(defaultTags()); - for (int i = 0; i < 2 * workItems; i += 2) { - matchers[3 + i] = - span() - .childOf(Worker.publisherParentId) - .operationName("addOne") - .tags(componentTrace(), defaultTags()); - matchers[4 + i] = - span() - .childOf(Worker.publisherParentId) - .operationName("addTwo") - .tags(componentTrace(), defaultTags()); - } - - assertTraces(trace(SORT_BY_START_TIME, matchers)); - } - - // --- Schedulers ---------------------------------------------------------- - - static List schedulerArgs() { - return Arrays.asList( - Arguments.of("new-thread", Schedulers.newThread()), - Arguments.of("computation", Schedulers.computation()), - Arguments.of("single", Schedulers.single()), - Arguments.of("trampoline", Schedulers.trampoline())); - } - - @ParameterizedTest(name = "Flowables produce the right number of results on ''{0}'' scheduler") - @MethodSource("schedulerArgs") - void schedulers(String schedulerName, Scheduler scheduler) { - List values = - Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) - .parallel() - .runOn(scheduler) - .flatMap( - num -> - Maybe.just(num.toString() + " on " + Thread.currentThread().getName()) - .toFlowable()) - .sequential() - .toList() - .blockingGet(); - - assertEquals(4, values.size()); - - // No trace-parent span is active while the chain is assembled, so the instrumentation must be - // non-intrusive: parallel scheduler hops must not synthesize any trace. Flushing makes sure - // any span that the instrumentation might have wrongly created on the scheduler threads is - // reported before we assert the writer is empty. - tracer.flush(); - assertEquals( - 0, - writer.getTraceCount(), - () -> "Unexpected traces emitted without active trace: " + writer); - } - - @ParameterizedTest(name = "Flowable propagates context on ''{0}'' scheduler") - @MethodSource("schedulerArgs") - void flowableParallelContextPropagation(String schedulerName, Scheduler scheduler) { - Worker.assemblePublisherUnderTrace( - () -> - Flowable.fromIterable(Arrays.asList(1, 2, 3, 4)) - .parallel() - .runOn(scheduler) - .flatMap(num -> Maybe.just(num).map(Worker::addOne).toFlowable()) - .sequential()); - - SpanMatcher[] matchers = new SpanMatcher[6]; - matchers[0] = - span() - .root() - .operationName("trace-parent") - .resourceName("trace-parent") - .tags(componentTrace(), defaultTags()); - matchers[1] = - span() - .id(Worker.publisherParentId) - .childOf(Worker.traceParentId) - .operationName("publisher-parent") - .resourceName("publisher-parent") - .tags(defaultTags()); - for (int i = 0; i < 4; i++) { - matchers[2 + i] = - span() - .childOf(Worker.publisherParentId) - .operationName("addOne") - .resourceName("addOne") - .tags(componentTrace(), defaultTags()); - } - - assertTraces(trace(SORT_BY_START_TIME, matchers)); - } - - // --- No spurious traces outside active trace -------------------------------- - - // Verifies that CaptureParentSpanAdvice and PropagateParentSpanAdvice are no-ops when there is - // no active trace: the instrumentation must not synthesize any spans of its own. - static List noSpuriousTracesArgs() { - return Arrays.asList( - Arguments.of( - "observable", - (Supplier) - () -> - Observable.fromIterable(Arrays.asList(1, 2, 3, 4)) - .map(i -> i + 1) - .toList() - .blockingGet()), - Arguments.of( - "single", (Supplier) () -> Single.just(1).map(i -> i + 1).blockingGet())); - } - - @ParameterizedTest(name = "No spurious traces for ''{0}'' assembled outside active trace") - @MethodSource("noSpuriousTracesArgs") - void noSpuriousTracesWhenAssembledOutsideTrace(String name, Supplier supplier) { - supplier.get(); - tracer.flush(); - assertEquals( - 0, - writer.getTraceCount(), - () -> "Unexpected traces emitted without active trace: " + writer); - } -} diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/SubscriptionTest.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/SubscriptionTest.java deleted file mode 100644 index d2f76d17a9b..00000000000 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/testdog/trace/instrumentation/rxjava3/SubscriptionTest.java +++ /dev/null @@ -1,162 +0,0 @@ -package testdog.trace.instrumentation.rxjava3; - -import static datadog.trace.agent.test.assertions.SpanMatcher.span; -import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; -import static datadog.trace.agent.test.assertions.TraceMatcher.trace; -import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; -import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; - -import datadog.trace.agent.test.AbstractInstrumentationTest; -import datadog.trace.bootstrap.instrumentation.api.AgentScope; -import datadog.trace.bootstrap.instrumentation.api.AgentSpan; -import io.reactivex.rxjava3.core.BackpressureStrategy; -import io.reactivex.rxjava3.core.Completable; -import io.reactivex.rxjava3.core.Flowable; -import io.reactivex.rxjava3.core.Maybe; -import io.reactivex.rxjava3.core.Observable; -import io.reactivex.rxjava3.core.Single; -import java.util.Random; -import java.util.concurrent.CountDownLatch; -import org.junit.jupiter.api.Test; - -class SubscriptionTest extends AbstractInstrumentationTest { - - @Test - void maybeSubscriptionPropagatesParentSpan() throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - - AgentSpan parent = startSpan("test", "parent"); - try (AgentScope scope = activateSpan(parent)) { - Maybe connection = Maybe.create(emitter -> emitter.onSuccess(new Connection())); - connection.subscribe( - c -> { - c.query(); - latch.countDown(); - }); - } finally { - parent.finish(); - } - latch.await(); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("parent"), - span().childOfPrevious().operationName("Connection.query"))); - } - - @Test - void singleSubscriptionPropagatesParentSpan() throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - - AgentSpan parent = startSpan("test", "parent"); - try (AgentScope scope = activateSpan(parent)) { - Single connection = Single.create(emitter -> emitter.onSuccess(new Connection())); - connection.subscribe( - c -> { - c.query(); - latch.countDown(); - }); - } finally { - parent.finish(); - } - latch.await(); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("parent"), - span().childOfPrevious().operationName("Connection.query"))); - } - - @Test - void completableSubscriptionPropagatesParentSpan() throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - - AgentSpan parent = startSpan("test", "parent"); - try (AgentScope scope = activateSpan(parent)) { - Completable action = Completable.create(emitter -> emitter.onComplete()); - action.subscribe( - () -> { - new Connection().query(); - latch.countDown(); - }); - } finally { - parent.finish(); - } - latch.await(); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("parent"), - span().childOfPrevious().operationName("Connection.query"))); - } - - @Test - void observableSubscriptionPropagatesParentSpan() throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - - AgentSpan parent = startSpan("test", "parent"); - try (AgentScope scope = activateSpan(parent)) { - Observable connection = - Observable.create( - emitter -> { - emitter.onNext(new Connection()); - emitter.onComplete(); - }); - connection.subscribe( - c -> { - c.query(); - latch.countDown(); - }); - } finally { - parent.finish(); - } - latch.await(); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("parent"), - span().childOfPrevious().operationName("Connection.query"))); - } - - @Test - void flowableSubscriptionPropagatesParentSpan() throws InterruptedException { - CountDownLatch latch = new CountDownLatch(1); - - AgentSpan parent = startSpan("test", "parent"); - try (AgentScope scope = activateSpan(parent)) { - Flowable connection = - Flowable.create( - emitter -> { - emitter.onNext(new Connection()); - emitter.onComplete(); - }, - BackpressureStrategy.BUFFER); - connection.subscribe( - c -> { - c.query(); - latch.countDown(); - }); - } finally { - parent.finish(); - } - latch.await(); - - assertTraces( - trace( - SORT_BY_START_TIME, - span().root().operationName("parent"), - span().childOfPrevious().operationName("Connection.query"))); - } - - static class Connection { - int query() { - AgentSpan span = startSpan("test", "Connection.query"); - span.finish(); - return new Random().nextInt(); - } - } -}