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..eb2577ff918 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 @@ -18,7 +18,6 @@ public final class CompletableInstrumentation implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { - @Override public String instrumentedType() { return "io.reactivex.rxjava3.core.Completable"; @@ -55,6 +54,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..ca81d856240 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 @@ -18,7 +18,6 @@ public final class FlowableInstrumentation implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { - @Override public String instrumentedType() { return "io.reactivex.rxjava3.core.Flowable"; @@ -54,6 +53,7 @@ public static ContextScope onSubscribe( 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..cbc647440c2 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 @@ -13,40 +13,39 @@ @AutoService(InstrumenterModule.class) public final class RxJavaModule extends InstrumenterModule.ContextTracking { public RxJavaModule() { - super("rxjava", "rxjava-3"); + super("rxjava"); } @Override public String[] helperClassNames() { return new String[] { - packageName + ".TracingCompletableObserver", + packageName + ".TracingObserver", packageName + ".TracingSubscriber", + packageName + ".TracingSingleObserver", packageName + ".TracingMaybeObserver", - packageName + ".TracingObserver", + packageName + ".TracingCompletableObserver", packageName + ".RxJavaAsyncResultExtension", - packageName + ".TracingSingleObserver", }; } @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.Observable", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Flowable", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Single", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Maybe", Context.class.getName()); + store.put("io.reactivex.rxjava3.core.Completable", Context.class.getName()); return store; } @Override public List typeInstrumentations() { return asList( - new CompletableInstrumentation(), + new ObservableInstrumentation(), new FlowableInstrumentation(), + new SingleInstrumentation(), new MaybeInstrumentation(), - new ObservableInstrumentation(), - new SingleInstrumentation()); + new CompletableInstrumentation()); } } 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..89557ca2dac 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 @@ -18,7 +18,6 @@ public final class SingleInstrumentation implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { - @Override public String instrumentedType() { return "io.reactivex.rxjava3.core.Single"; @@ -53,6 +52,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/TracingCompletableObserver.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingCompletableObserver.java index 8a0dd7254e1..dd83c901c30 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingCompletableObserver.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingCompletableObserver.java @@ -4,15 +4,17 @@ import datadog.context.ContextScope; import io.reactivex.rxjava3.core.CompletableObserver; 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. */ +/** + * Wrapper that makes sure spans from completable-observer events treat the captured span as their + * parent. + */ public final class TracingCompletableObserver implements CompletableObserver { private final CompletableObserver observer; private final Context parentContext; public TracingCompletableObserver( - @Nonnull final CompletableObserver observer, @Nonnull final Context parentContext) { + final CompletableObserver 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/TracingMaybeObserver.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingMaybeObserver.java index 0cbf34c61e4..2f7700f484d 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingMaybeObserver.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingMaybeObserver.java @@ -4,15 +4,15 @@ import datadog.context.ContextScope; import io.reactivex.rxjava3.core.MaybeObserver; 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. */ +/** + * Wrapper that makes sure spans from maybe-observer events treat the captured span as their parent. + */ public final class TracingMaybeObserver implements MaybeObserver { private final MaybeObserver observer; private final Context parentContext; - public TracingMaybeObserver( - @Nonnull final MaybeObserver observer, @Nonnull final Context parentContext) { + public TracingMaybeObserver(final MaybeObserver 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/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/TracingSingleObserver.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSingleObserver.java index 3e05d1124bc..4f658cf21f2 100644 --- a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSingleObserver.java +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/TracingSingleObserver.java @@ -4,15 +4,16 @@ import datadog.context.ContextScope; import io.reactivex.rxjava3.core.SingleObserver; 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. */ +/** + * Wrapper that makes sure spans from single-observer events treat the captured span as their + * parent. + */ public final class TracingSingleObserver implements SingleObserver { private final SingleObserver observer; private final Context parentContext; - public TracingSingleObserver( - @Nonnull final SingleObserver observer, @Nonnull final Context parentContext) { + public TracingSingleObserver(final SingleObserver 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..861e268ef73 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 @@ -18,7 +18,7 @@ public TracingSubscriber( } @Override - public void onSubscribe(final Subscription subscription) { + public void onSubscribe(@Nonnull final Subscription subscription) { subscriber.onSubscribe(subscription); } 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 index 163f5fb4b69..32bde397d30 100644 --- 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 @@ -10,6 +10,7 @@ import io.reactivex.rxjava3.core.Single; import java.util.concurrent.CountDownLatch; +/** Helper methods annotated with {@code @WithSpan} for testing async result extensions. */ public class RxJava3TracedMethods { @WithSpan public static Completable traceAsyncCompletable(CountDownLatch latch) { @@ -100,31 +101,6 @@ public static Flowable traceAsyncFailingFlowable( }); } - @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)) { diff --git a/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3ResultExtensionTest.java b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3ResultExtensionTest.java new file mode 100644 index 00000000000..6e0e1e49bd1 --- /dev/null +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3ResultExtensionTest.java @@ -0,0 +1,190 @@ +package datadog.trace.instrumentation.rxjava3; + +import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; +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.Matchers; +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 java.util.concurrent.CountDownLatch; +import java.util.regex.Pattern; +import java.util.stream.Stream; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Tests for RxJava 3 async result extension with {@code @WithSpan} annotated methods. + * + *

Verifies that when a method annotated with {@code @WithSpan} returns an RxJava 3 reactive + * type, the span is completed when the reactive type completes (success, error, or cancellation), + * not when the method returns. + */ +@WithConfig(key = "dd.trace.otel.enabled", value = "true") +@WithConfig(key = "dd.integration.opentelemetry-annotations-1.20.enabled", value = "true") +class RxJava3ResultExtensionTest extends AbstractInstrumentationTest { + + // --------------------------------------------------------------------------- + // Successful async completion: span finishes when reactive type completes + // --------------------------------------------------------------------------- + + static Stream asyncSuccessTestCases() { + return Stream.of( + Arguments.of("Completable", "blockingAwait"), + Arguments.of("Maybe", "blockingGet"), + Arguments.of("Single", "blockingGet"), + Arguments.of("Observable", "blockingLast"), + Arguments.of("Flowable", "blockingLast")); + } + + @ParameterizedTest(name = "test WithSpan annotated async method {0}") + @MethodSource("asyncSuccessTestCases") + void asyncSuccessTest(String type, String operation) throws Exception { + CountDownLatch latch = new CountDownLatch(1); + String methodName = "traceAsync" + type; + Object asyncType = invokeTracedMethod(methodName, latch); + + // No spans should be written yet — the reactive type has not completed + assertEquals(0, writer.size()); + + latch.countDown(); + invokeBlockingOp(asyncType, operation); + + String expectedOpName = "RxJava3TracedMethods." + methodName; + String expectedResName = "RxJava3TracedMethods." + methodName; + assertTraces( + trace( + SpanMatcher.span() + .operationName(Pattern.compile(Pattern.quote(expectedOpName))) + .resourceName(Pattern.compile(Pattern.quote(expectedResName))) + .root() + .tags( + defaultTags(), + TagsMatcher.tag(Tags.COMPONENT, Matchers.matches("opentelemetry")), + TagsMatcher.tag(Tags.SPAN_KIND, Matchers.matches("internal"))))); + } + + // --------------------------------------------------------------------------- + // Failing async completion: span finishes with error when reactive type errors + // --------------------------------------------------------------------------- + + static Stream asyncFailureTestCases() { + return Stream.of( + Arguments.of("Completable", "blockingAwait"), + Arguments.of("Maybe", "blockingGet"), + Arguments.of("Single", "blockingGet"), + Arguments.of("Observable", "blockingLast"), + Arguments.of("Flowable", "blockingLast")); + } + + @ParameterizedTest(name = "test WithSpan annotated async method failing {0}") + @MethodSource("asyncFailureTestCases") + void asyncFailureTest(String type, String operation) throws Exception { + CountDownLatch latch = new CountDownLatch(1); + IllegalStateException expectedException = new IllegalStateException("Test exception"); + String methodName = "traceAsyncFailing" + type; + Object asyncType = invokeTracedMethodFailing(methodName, latch, expectedException); + + assertEquals(0, writer.size()); + + latch.countDown(); + assertThrows(RuntimeException.class, () -> invokeBlockingOp(asyncType, operation)); + + String expectedName = "RxJava3TracedMethods." + methodName; + assertTraces( + trace( + SpanMatcher.span() + .operationName(Pattern.compile(Pattern.quote(expectedName))) + .resourceName(Pattern.compile(Pattern.quote(expectedName))) + .root() + .error() + .tags( + defaultTags(), + TagsMatcher.tag(Tags.COMPONENT, Matchers.matches("opentelemetry")), + TagsMatcher.tag(Tags.SPAN_KIND, Matchers.matches("internal")), + TagsMatcher.error(expectedException)))); + } + + // --------------------------------------------------------------------------- + // Cancelled async: span finishes when subscriber disposes + // --------------------------------------------------------------------------- + + static Stream asyncCancelTestCases() { + return Stream.of( + Arguments.of("Completable"), + Arguments.of("Maybe"), + Arguments.of("Single"), + Arguments.of("Observable"), + Arguments.of("Flowable")); + } + + @ParameterizedTest(name = "test WithSpan annotated async method cancelled {0}") + @MethodSource("asyncCancelTestCases") + void asyncCancelTest(String type) throws Exception { + CountDownLatch latch = new CountDownLatch(1); + String methodName = "traceAsync" + type; + Object asyncType = invokeTracedMethod(methodName, latch); + + assertEquals(0, writer.size()); + + latch.countDown(); + invokeSubscribeAndDispose(asyncType); + + String expectedOpName = "RxJava3TracedMethods." + methodName; + String expectedResName = "RxJava3TracedMethods." + methodName; + assertTraces( + trace( + SpanMatcher.span() + .operationName(Pattern.compile(Pattern.quote(expectedOpName))) + .resourceName(Pattern.compile(Pattern.quote(expectedResName))) + .root() + .tags( + defaultTags(), + TagsMatcher.tag(Tags.COMPONENT, Matchers.matches("opentelemetry")), + TagsMatcher.tag(Tags.SPAN_KIND, Matchers.matches("internal"))))); + } + + // --------------------------------------------------------------------------- + // Reflective helper methods to invoke traced methods by name + // --------------------------------------------------------------------------- + + private static Object invokeTracedMethod(String methodName, CountDownLatch latch) + throws Exception { + return RxJava3TracedMethods.class + .getMethod(methodName, CountDownLatch.class) + .invoke(null, latch); + } + + private static Object invokeTracedMethodFailing( + String methodName, CountDownLatch latch, Exception exception) throws Exception { + return RxJava3TracedMethods.class + .getMethod(methodName, CountDownLatch.class, Exception.class) + .invoke(null, latch, exception); + } + + private static void invokeBlockingOp(Object asyncType, String operation) throws Exception { + try { + asyncType.getClass().getMethod(operation).invoke(asyncType); + } catch (java.lang.reflect.InvocationTargetException e) { + Throwable cause = e.getCause(); + if (cause instanceof RuntimeException) { + throw (RuntimeException) cause; + } else if (cause instanceof Error) { + throw (Error) cause; + } else { + throw new RuntimeException(cause); + } + } + } + + private static void invokeSubscribeAndDispose(Object asyncType) throws Exception { + Object disposable = asyncType.getClass().getMethod("subscribe").invoke(asyncType); + disposable.getClass().getMethod("dispose").invoke(disposable); + } +} 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..f11467c4506 --- /dev/null +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3SubscriptionTest.java @@ -0,0 +1,419 @@ +package datadog.trace.instrumentation.rxjava3; + +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.agent.test.utils.TraceUtils.runUnderTrace; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.assertions.SpanMatcher; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.AgentTracer; +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 io.reactivex.rxjava3.schedulers.Schedulers; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import org.junit.jupiter.api.Test; + +/** + * Context-propagation tests for RxJava 3 subscription. + * + *

Verifies that when a reactive type is subscribed to under an active span, the trace context is + * propagated into the subscriber callbacks so that child spans created inside those callbacks + * become children of the parent span. + */ +class RxJava3SubscriptionTest extends AbstractInstrumentationTest { + + @Test + void maybeSubscriptionPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Maybe connection = + Maybe.create(emitter -> emitter.onSuccess(new Connection())); + connection.subscribe( + conn -> { + conn.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void observableSubscriptionPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Observable.just(new Connection()) + .subscribe( + conn -> { + conn.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void singleSubscriptionPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Single.just(new Connection()) + .subscribe( + conn -> { + conn.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void flowableSubscriptionPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Flowable.just(new Connection()) + .subscribe( + conn -> { + conn.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void completableSubscriptionPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Completable.fromAction( + () -> { + new Connection().query(); + latch.countDown(); + }) + .subscribe(); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void contextPropagatesAcrossThreadBoundary() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Maybe.create(emitter -> emitter.onSuccess(new Connection())) + .subscribeOn(Schedulers.io()) + .subscribe( + conn -> { + ((Connection) conn).query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void multipleSubscribersCaptureOwnParent() throws Exception { + CountDownLatch latch = new CountDownLatch(2); + + runUnderTrace( + "parent", + () -> { + Maybe maybe = Maybe.create(emitter -> emitter.onSuccess(new Connection())); + + maybe.subscribe( + conn -> { + conn.query(); + latch.countDown(); + }); + + maybe.subscribe( + conn -> { + conn.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void observableOnErrorPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Observable.error(new RuntimeException("boom")) + .subscribe( + item -> {}, + error -> { + Connection.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void flowableOnErrorPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Flowable.error(new RuntimeException("boom")) + .subscribe( + item -> {}, + error -> { + Connection.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void singleOnErrorPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Single.error(new RuntimeException("boom")) + .subscribe( + item -> {}, + error -> { + Connection.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void maybeOnErrorPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Maybe.error(new RuntimeException("boom")) + .subscribe( + item -> {}, + error -> { + Connection.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + @Test + void completableOnErrorPropagatesContext() throws Exception { + CountDownLatch latch = new CountDownLatch(1); + runUnderTrace( + "parent", + () -> { + Completable.error(new RuntimeException("boom")) + .subscribe( + () -> {}, + error -> { + Connection.query(); + latch.countDown(); + }); + return null; + }); + latch.await(5, TimeUnit.SECONDS); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("parent") + .resourceName("parent") + .root() + .tags(defaultTags()), + SpanMatcher.span() + .operationName("Connection.query") + .resourceName("Connection.query") + .childOfIndex(0))); + } + + /** Simple helper that creates a traced span when its {@code query()} method is called. */ + static class Connection { + static int query() { + AgentSpan span = AgentTracer.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..0a28cd98e4e --- /dev/null +++ b/dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/test/java/datadog/trace/instrumentation/rxjava3/RxJava3Test.java @@ -0,0 +1,546 @@ +package datadog.trace.instrumentation.rxjava3; + +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.agent.test.utils.TraceUtils.runUnderTrace; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +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.Matchers; +import datadog.trace.agent.test.assertions.SpanMatcher; +import datadog.trace.agent.test.assertions.TagsMatcher; +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.schedulers.Schedulers; +import java.util.Arrays; +import java.util.List; +import java.util.function.Function; +import java.util.stream.Stream; +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; + +/** + * Context-propagation tests for RxJava 3. + * + *

RxJava instrumentation does NOT create spans itself — it propagates the active trace context + * across async boundaries so that spans started inside reactive callbacks become children of the + * span that was active when the reactive chain was subscribed to. + */ +class RxJava3Test extends AbstractInstrumentationTest { + + // --------------------------------------------------------------------------- + // Helper traced methods — each creates a child span under the current context + // --------------------------------------------------------------------------- + + static int addOneFunc(int i) { + AgentSpan span = startSpan("test", "addOne"); + span.setResourceName("addOne"); + span.setTag("component", "trace"); + AgentScope scope = activateSpan(span); + try { + return i + 1; + } finally { + scope.close(); + span.finish(); + } + } + + static int addTwoFunc(int i) { + AgentSpan span = startSpan("test", "addTwo"); + span.setResourceName("addTwo"); + span.setTag("component", "trace"); + AgentScope scope = activateSpan(span); + try { + return i + 2; + } finally { + scope.close(); + span.finish(); + } + } + + private static final Function ADD_ONE = RxJava3Test::addOneFunc; + private static final Function ADD_TWO = RxJava3Test::addTwoFunc; + + // --------------------------------------------------------------------------- + // Publisher test: verify child spans are parented under the trace-parent + // --------------------------------------------------------------------------- + + static Stream publisherTestCases() { + return Stream.of( + Arguments.of( + "basic maybe", 2, 1, (CallableSupplier) () -> Maybe.just(1).map(ADD_ONE::apply)), + Arguments.of( + "two operations maybe", + 4, + 2, + (CallableSupplier) () -> Maybe.just(2).map(ADD_ONE::apply).map(ADD_ONE::apply)), + Arguments.of( + "delayed maybe", + 4, + 1, + (CallableSupplier) () -> Maybe.just(3).delay(100, MILLISECONDS).map(ADD_ONE::apply)), + Arguments.of( + "delayed twice maybe", + 6, + 2, + (CallableSupplier) + () -> + Maybe.just(4) + .delay(100, MILLISECONDS) + .map(ADD_ONE::apply) + .delay(100, MILLISECONDS) + .map(ADD_ONE::apply)), + Arguments.of( + "basic flowable", + null, + 2, + (CallableSupplier) + () -> Flowable.fromIterable(Arrays.asList(5, 6)).map(ADD_ONE::apply)), + Arguments.of( + "two operations flowable", + null, + 4, + (CallableSupplier) + () -> + Flowable.fromIterable(Arrays.asList(6, 7)) + .map(ADD_ONE::apply) + .map(ADD_ONE::apply)), + Arguments.of( + "delayed flowable", + null, + 2, + (CallableSupplier) + () -> + Flowable.fromIterable(Arrays.asList(7, 8)) + .delay(100, MILLISECONDS) + .map(ADD_ONE::apply)), + Arguments.of( + "delayed twice flowable", + null, + 4, + (CallableSupplier) + () -> + Flowable.fromIterable(Arrays.asList(8, 9)) + .delay(100, MILLISECONDS) + .map(ADD_ONE::apply) + .delay(100, MILLISECONDS) + .map(ADD_ONE::apply)), + Arguments.of( + "maybe from callable", + 12, + 2, + (CallableSupplier) () -> Maybe.fromCallable(() -> addOneFunc(10)).map(ADD_ONE::apply))); + } + + @ParameterizedTest(name = "Publisher ''{0}'' test") + @MethodSource("publisherTestCases") + void publisherTest( + String name, Object expected, int workSpans, CallableSupplier publisherSupplier) + throws Exception { + Object result = assemblePublisherUnderTrace(publisherSupplier); + + SpanMatcher[] spanMatchers = new SpanMatcher[workSpans + 2]; + spanMatchers[0] = + SpanMatcher.span().operationName("trace-parent").resourceName("trace-parent").root(); + spanMatchers[1] = + SpanMatcher.span() + .operationName("publisher-parent") + .resourceName("publisher-parent") + .childOfIndex(0); + for (int i = 0; i < workSpans; i++) { + spanMatchers[i + 2] = + SpanMatcher.span() + .operationName("addOne") + .resourceName("addOne") + .childOfIndex(1) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))); + } + assertTraces(trace(SORT_BY_START_TIME, spanMatchers)); + } + + // --------------------------------------------------------------------------- + // Publisher error: errors propagate up but do NOT set error on the reactive span + // --------------------------------------------------------------------------- + + static Stream publisherErrorTestCases() { + return Stream.of( + Arguments.of( + "maybe", (CallableSupplier) () -> Maybe.error(new RuntimeException("test exception"))), + Arguments.of( + "flowable", + (CallableSupplier) () -> Flowable.error(new RuntimeException("test exception")))); + } + + @ParameterizedTest(name = "Publisher error ''{0}'' test") + @MethodSource("publisherErrorTestCases") + void publisherErrorTest(String name, CallableSupplier publisherSupplier) { + assertThrows(RuntimeException.class, () -> assemblePublisherUnderTrace(publisherSupplier)); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span() + .operationName("trace-parent") + .resourceName("trace-parent") + .root() + .error() + .tags(defaultTags(), TagsMatcher.error(RuntimeException.class, "test exception")), + SpanMatcher.span() + .operationName("publisher-parent") + .resourceName("publisher-parent") + .childOfIndex(0))); + } + + // --------------------------------------------------------------------------- + // Publisher step error: error in map step after a successful step + // --------------------------------------------------------------------------- + + static Stream publisherStepErrorTestCases() { + return Stream.of( + Arguments.of( + "basic maybe failure", + 1, + (CallableSupplier) + () -> + Maybe.just(1) + .map(ADD_ONE::apply) + .map( + i -> { + throw new RuntimeException("test exception"); + })), + Arguments.of( + "basic flowable failure", + 1, + (CallableSupplier) + () -> + Flowable.fromIterable(Arrays.asList(5, 6)) + .map(ADD_ONE::apply) + .map( + i -> { + throw new RuntimeException("test exception"); + }))); + } + + @ParameterizedTest(name = "Publisher step ''{0}'' test") + @MethodSource("publisherStepErrorTestCases") + void publisherStepErrorTest(String name, int workSpans, CallableSupplier publisherSupplier) { + assertThrows(RuntimeException.class, () -> assemblePublisherUnderTrace(publisherSupplier)); + + SpanMatcher[] spanMatchers = new SpanMatcher[workSpans + 2]; + spanMatchers[0] = + SpanMatcher.span() + .operationName("trace-parent") + .resourceName("trace-parent") + .root() + .error() + .tags(defaultTags(), TagsMatcher.error(RuntimeException.class, "test exception")); + spanMatchers[1] = + SpanMatcher.span() + .operationName("publisher-parent") + .resourceName("publisher-parent") + .childOfIndex(0); + for (int i = 0; i < workSpans; i++) { + spanMatchers[i + 2] = + SpanMatcher.span() + .operationName("addOne") + .resourceName("addOne") + .childOfIndex(1) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))); + } + assertTraces(trace(SORT_BY_START_TIME, spanMatchers)); + } + + // --------------------------------------------------------------------------- + // Publisher cancel: cancelled subscription still produces correct trace + // --------------------------------------------------------------------------- + + static Stream publisherCancelTestCases() { + return Stream.of( + Arguments.of("basic maybe", (CallableSupplier) () -> Maybe.just(1)), + Arguments.of( + "basic flowable", (CallableSupplier) () -> Flowable.fromIterable(Arrays.asList(5, 6)))); + } + + @ParameterizedTest(name = "Publisher ''{0}'' cancel") + @MethodSource("publisherCancelTestCases") + void publisherCancelTest(String name, CallableSupplier publisherSupplier) throws Exception { + cancelUnderTrace(publisherSupplier); + + assertTraces( + trace( + SpanMatcher.span().operationName("trace-parent").resourceName("trace-parent").root(), + SpanMatcher.span() + .operationName("publisher-parent") + .resourceName("publisher-parent") + .childOfIndex(0))); + } + + // --------------------------------------------------------------------------- + // Chain spans: correct parent across concat operations + // --------------------------------------------------------------------------- + + static Stream chainSpansTestCases() { + return Stream.of( + Arguments.of( + "basic maybe", + 3, + (CallableSupplier) + () -> + Maybe.just(1) + .map(ADD_ONE::apply) + .map(ADD_ONE::apply) + .concatWith(Maybe.just(1).map(ADD_ONE::apply))), + Arguments.of( + "basic flowable", + 5, + (CallableSupplier) + () -> + Flowable.fromIterable(Arrays.asList(5, 6)) + .map(ADD_ONE::apply) + .map(ADD_ONE::apply) + .concatWith(Maybe.just(1).map(ADD_ONE::apply).toFlowable()))); + } + + @ParameterizedTest(name = "Publisher chain spans have the correct parent for ''{0}''") + @MethodSource("chainSpansTestCases") + void chainSpansTest(String name, int workSpans, CallableSupplier publisherSupplier) + throws Exception { + assemblePublisherUnderTrace(publisherSupplier); + + SpanMatcher[] spanMatchers = new SpanMatcher[workSpans + 2]; + spanMatchers[0] = + SpanMatcher.span().operationName("trace-parent").resourceName("trace-parent").root(); + spanMatchers[1] = + SpanMatcher.span() + .operationName("publisher-parent") + .resourceName("publisher-parent") + .childOfIndex(0); + for (int i = 0; i < workSpans; i++) { + spanMatchers[i + 2] = + SpanMatcher.span() + .operationName("addOne") + .resourceName("addOne") + .childOfIndex(1) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))); + } + assertTraces(trace(SORT_BY_START_TIME, spanMatchers)); + } + + // --------------------------------------------------------------------------- + // Subscription-time parent: spans inherit the parent active at subscribe time + // --------------------------------------------------------------------------- + + @Test + void publisherChainSpansHaveCorrectParentFromSubscriptionTime() throws Exception { + Maybe maybe = Maybe.just(42).map(ADD_ONE::apply).map(ADD_TWO::apply); + + runUnderTrace( + "trace-parent", + () -> { + maybe.blockingGet(); + return null; + }); + + assertTraces( + trace( + SORT_BY_START_TIME, + SpanMatcher.span().operationName("trace-parent").resourceName("trace-parent").root(), + SpanMatcher.span() + .operationName("addOne") + .childOfIndex(0) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))), + SpanMatcher.span() + .operationName("addTwo") + .childOfIndex(0) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))))); + } + + // --------------------------------------------------------------------------- + // Intermediate span: spans get the right parent when an intermediate scope + // is active between publisher creation and subscribe + // --------------------------------------------------------------------------- + + static Stream intermediateSpanTestCases() { + return Stream.of( + Arguments.of("basic maybe", 1, (CallableSupplier) () -> Maybe.just(1).map(ADD_ONE::apply)), + Arguments.of( + "basic flowable", + 2, + (CallableSupplier) + () -> Flowable.fromIterable(Arrays.asList(1, 2)).map(ADD_ONE::apply))); + } + + @ParameterizedTest( + name = "Publisher chain spans have the correct parents from subscription time ''{0}''") + @MethodSource("intermediateSpanTestCases") + void intermediateSpanTest(String name, int workItems, CallableSupplier publisherSupplier) + throws Exception { + assemblePublisherWithIntermediateSpan(publisherSupplier, workItems); + + int totalSpans = 3 + 2 * workItems; + SpanMatcher[] spanMatchers = new SpanMatcher[totalSpans]; + spanMatchers[0] = + SpanMatcher.span().operationName("trace-parent").resourceName("trace-parent").root(); + spanMatchers[1] = + SpanMatcher.span() + .operationName("publisher-parent") + .resourceName("publisher-parent") + .childOfIndex(0); + spanMatchers[2] = SpanMatcher.span().operationName("intermediate").childOfIndex(1); + + for (int i = 0; i < 2 * workItems; i += 2) { + spanMatchers[3 + i] = + SpanMatcher.span() + .operationName("addOne") + .childOfIndex(1) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))); + spanMatchers[3 + i + 1] = + SpanMatcher.span() + .operationName("addTwo") + .childOfIndex(1) + .tags(defaultTags(), TagsMatcher.tag("component", Matchers.is("trace"))); + } + assertTraces(trace(SORT_BY_START_TIME, spanMatchers)); + } + + // --------------------------------------------------------------------------- + // Scheduler test: flowable produces correct number of results on each scheduler + // --------------------------------------------------------------------------- + + static Stream schedulerTestCases() { + return Stream.of( + 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("schedulerTestCases") + void schedulerTest(String schedulerName, io.reactivex.rxjava3.core.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()); + } + + // --------------------------------------------------------------------------- + // Helper methods + // --------------------------------------------------------------------------- + + private Object assemblePublisherUnderTrace(CallableSupplier publisherSupplier) throws Exception { + return runUnderTrace( + "trace-parent", + () -> { + AgentSpan span = startSpan("test", "publisher-parent"); + AgentScope scope = activateSpan(span); + + Object publisher = publisherSupplier.call(); + try { + if (publisher instanceof Maybe) { + return ((Maybe) publisher).blockingGet(); + } else if (publisher instanceof Flowable) { + return ((Flowable) publisher).toList().blockingGet().toArray(new Integer[0]); + } + throw new RuntimeException("Unknown publisher: " + publisher); + } finally { + span.finish(); + scope.close(); + } + }); + } + + private void cancelUnderTrace(CallableSupplier publisherSupplier) throws Exception { + runUnderTrace( + "trace-parent", + () -> { + AgentSpan span = startSpan("test", "publisher-parent"); + AgentScope scope = activateSpan(span); + + Object publisher = publisherSupplier.call(); + if (publisher instanceof Maybe) { + publisher = ((Maybe) publisher).toFlowable(); + } + + ((Flowable) publisher) + .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(); + return null; + }); + } + + @SuppressWarnings("unchecked") + private void assemblePublisherWithIntermediateSpan( + CallableSupplier publisherSupplier, int workItems) throws Exception { + runUnderTrace( + "trace-parent", + () -> { + AgentSpan parentSpan = startSpan("test", "publisher-parent"); + AgentScope parentScope = activateSpan(parentSpan); + + Object publisher = publisherSupplier.call(); + + AgentSpan intermediate = startSpan("test", "intermediate"); + AgentScope intermediateScope = activateSpan(intermediate); + try { + if (publisher instanceof Maybe) { + ((Maybe) publisher).map(ADD_TWO::apply).blockingGet(); + } else if (publisher instanceof Flowable) { + ((Flowable) publisher).map(ADD_TWO::apply).toList().blockingGet(); + } else { + throw new IllegalStateException("Unknown publisher type"); + } + } finally { + intermediate.finish(); + intermediateScope.close(); + } + + parentSpan.finish(); + parentScope.close(); + return null; + }); + } + + /** Functional interface that can throw checked exceptions. */ + @FunctionalInterface + interface CallableSupplier { + Object call() throws Exception; + } +} 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(); - } - } -}