From dbf3d6b2c71b16fb9f17225d174758e4f8a50dc6 Mon Sep 17 00:00:00 2001 From: Jordan Wong Date: Tue, 14 Jul 2026 07:57:55 -0400 Subject: [PATCH 1/4] eval: blind regeneration of reactor-core-3.1 against PR #11927 draft skill Automated integration-authoring pass on reactor-core-3.1, produced against the draft skill in #11927. Reference for #11927 reviewers to see concrete output the current skill draft produces. Not intended to be merged as-is. Base: master @ 05671ce3b0 (rebased from original 2e1d2b98d7; the original branch inherited a stash-merge revert of #11852 during eval branch setup that dropped DD_TRACE_ROBOLECTRIC_ENABLED from metadata plus 172 unrelated files. This rebased version cherry-picks ONLY the reactor-core-3.1 module output onto current master, preserving all unrelated upstream changes.) Cost: \$13, reviewer approved with 0 todos remaining. Diff scope: dd-java-agent/instrumentation/reactor-core-3.1/ only. Metadata: no change (all reactor entries already present on master). Note: the toolkit reorganized the package layout from 'datadog.trace.instrumentation.reactor.core' (master) to 'datadog.trace.instrumentation.reactorcore' (eval). This is a regen-preservation Rule #2 concern; documented for reviewer consideration. --- .../reactor-core-3.1/NOTICE.txt | 17 - .../reactor-core-3.1/build.gradle | 20 +- .../reactor-core-3.1/gradle.lockfile | 141 ----- .../ReactorAsyncResultExtensionTest.groovy | 206 ------ .../groovy/ReactorCoreTest.groovy | 598 ------------------ .../groovy/SubscriptionTest.groovy | 126 ---- .../annotatedsample/ReactorTracedMethods.java | 75 --- .../BlockingPublisherInstrumentation.java | 55 -- ...ntextWritingSubscriberInstrumentation.java | 71 --- .../core/CorePublisherInstrumentation.java | 73 --- .../OptimizableOperatorInstrumentation.java | 62 -- .../reactor/core/ReactorContextBridge.java | 139 ---- .../reactor/core/ReactorCoreModule.java | 62 -- .../reactorcore/FluxInstrumentation.java | 70 ++ .../reactorcore/MonoInstrumentation.java | 72 +++ .../ReactorAsyncResultExtension.java | 49 ++ .../reactorcore/ReactorCoreModule.java | 39 ++ .../reactorcore/TracingCoreSubscriber.java | 43 ++ .../ReactorAsyncResultExtensionTest.groovy | 205 ------ .../src/test/groovy/ReactorCoreTest.groovy | 598 ------------------ .../src/test/groovy/SubscriptionTest.groovy | 214 ------- .../annotatedsample/ReactorTracedMethods.java | 75 --- .../reactorcore/ReactorCoreTest.java | 587 +++++++++++++++++ .../reactorcore/SubscriptionTest.java | 130 ++++ 24 files changed, 994 insertions(+), 2733 deletions(-) delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/NOTICE.txt delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/gradle.lockfile delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorAsyncResultExtensionTest.groovy delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorCoreTest.groovy delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/SubscriptionTest.groovy delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/java/annotatedsample/ReactorTracedMethods.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/BlockingPublisherInstrumentation.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ContextWritingSubscriberInstrumentation.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/CorePublisherInstrumentation.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/OptimizableOperatorInstrumentation.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorContextBridge.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorCoreModule.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorAsyncResultExtension.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/TracingCoreSubscriber.java delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorAsyncResultExtensionTest.groovy delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorCoreTest.groovy delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/SubscriptionTest.groovy delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/annotatedsample/ReactorTracedMethods.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/SubscriptionTest.java diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/NOTICE.txt b/dd-java-agent/instrumentation/reactor-core-3.1/NOTICE.txt deleted file mode 100644 index eab0fa0a007..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/NOTICE.txt +++ /dev/null @@ -1,17 +0,0 @@ -This product contains a modified part of OpenTracing: - - * License: - -Copyright 2018 The OpenTracing Authors - -Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except -in compliance with the License. You may obtain a copy of the License at - -http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software distributed under the License -is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express -or implied. See the License for the specific language governing permissions and limitations under -the License. - - * Homepage: https://github.com/opentracing-contrib/java-reactor diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/build.gradle b/dd-java-agent/instrumentation/reactor-core-3.1/build.gradle index e40cffa6d79..20358684ba7 100644 --- a/dd-java-agent/instrumentation/reactor-core-3.1/build.gradle +++ b/dd-java-agent/instrumentation/reactor-core-3.1/build.gradle @@ -1,4 +1,3 @@ - muzzle { pass { group = "io.projectreactor" @@ -9,24 +8,13 @@ muzzle { apply from: "$rootDir/gradle/java.gradle" -addTestSuite('latestDepTest') +addTestSuiteForDir('latestDepTest', 'test') dependencies { + compileOnly group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.0' compileOnly group: 'io.projectreactor', name: 'reactor-core', version: '3.1.0.RELEASE' - - testImplementation project(':dd-java-agent:instrumentation:datadog:tracing:trace-annotation') - testImplementation project(':dd-java-agent:instrumentation:reactive-streams-1.0') - testImplementation project(path: ':dd-java-agent:agent-otel:otel-bootstrap', configuration: 'shadow') - testImplementation project(':dd-java-agent:instrumentation:opentelemetry:opentelemetry-1.4') - testImplementation project(':dd-java-agent:instrumentation:opentelemetry:opentelemetry-annotations-1.20') - testImplementation project (':dd-java-agent:instrumentation:opentracing:opentracing-0.32') - testImplementation group: 'io.projectreactor', name: 'reactor-core', version: '3.1.0.RELEASE' - testImplementation group: 'io.opentelemetry.instrumentation', name: 'opentelemetry-instrumentation-annotations', version: '1.28.0' - testImplementation group: 'io.opentracing', name: 'opentracing-util', version: '0.32.0' - - latestDepTestImplementation group: 'io.projectreactor', name: 'reactor-core', version: '+' - // Looks like later versions on reactor need this dependency for some reason even though it is marked as optional. - latestDepTestImplementation group: 'io.micrometer', name: 'micrometer-core', version: '1.+' + testImplementation project(':dd-java-agent:instrumentation:datadog:tracing:trace-annotation') + latestDepTestImplementation group: 'io.projectreactor', name: 'reactor-core', version: '3.+' } diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/gradle.lockfile b/dd-java-agent/instrumentation/reactor-core-3.1/gradle.lockfile deleted file mode 100644 index 60374c20080..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/gradle.lockfile +++ /dev/null @@ -1,141 +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:reactor-core-3.1: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.micrometer:micrometer-commons:1.17.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath -io.micrometer:micrometer-core:1.17.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath -io.micrometer:micrometer-observation:1.17.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath -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.opentracing:opentracing-api:0.32.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.opentracing:opentracing-noop:0.32.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.opentracing:opentracing-util:0.32.0=latestDepTestCompileClasspath,latestDepTestRuntimeClasspath,testCompileClasspath,testRuntimeClasspath -io.projectreactor:reactor-core:3.1.0.RELEASE=compileClasspath,testCompileClasspath,testRuntimeClasspath -io.projectreactor:reactor-core:3.8.6=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.hdrhistogram:HdrHistogram:2.2.2=latestDepTestRuntimeClasspath -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.1=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/reactor-core-3.1/src/latestDepTest/groovy/ReactorAsyncResultExtensionTest.groovy b/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorAsyncResultExtensionTest.groovy deleted file mode 100644 index 92b0b42f720..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorAsyncResultExtensionTest.groovy +++ /dev/null @@ -1,206 +0,0 @@ -import annotatedsample.ReactorTracedMethods -import com.google.common.util.concurrent.ListeningExecutorService -import com.google.common.util.concurrent.MoreExecutors -import datadog.trace.agent.test.InstrumentationSpecification -import datadog.trace.bootstrap.instrumentation.api.Tags -import spock.lang.Shared - -import java.util.concurrent.CountDownLatch -import java.util.concurrent.Executors - -class ReactorAsyncResultExtensionTest extends InstrumentationSpecification { - @Override - void configurePreAgent() { - super.configurePreAgent() - - injectSysConfig("dd.trace.otel.enabled", "true") - injectSysConfig("dd.integration.opentelemetry-annotations-1.20.enabled", "true") - } - - @Shared - ListeningExecutorService executor - - def setupSpec() { - this.executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()) - } - - def cleanupSpec() { - this.executor.shutdownNow() - } - - def "test WithSpan annotated async method (Mono)"() { - setup: - def latch = new CountDownLatch(1) - def mono = ReactorTracedMethods.traceAsyncMono(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - mono.block() - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncMono" - operationName "ReactorTracedMethods.traceAsyncMono" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } - - def "test WithSpan annotated async method (failing Mono)"() { - setup: - def latch = new CountDownLatch(1) - def expectedException = new IllegalStateException("Test exception") - def mono = ReactorTracedMethods.traceAsyncFailingMono(latch, expectedException) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - mono.block() - - then: - thrown(IllegalStateException) - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFailingMono" - operationName "ReactorTracedMethods.traceAsyncFailingMono" - errored true - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - errorTags(expectedException) - } - } - } - } - } - - def "test WithSpan annotated async method (cancelled Mono)"() { - setup: - def latch = new CountDownLatch(1) - def mono = ReactorTracedMethods.traceAsyncMono(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - mono.subscribe(new ReactorTracedMethods.CancelSubscriber<>()) - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncMono" - operationName "ReactorTracedMethods.traceAsyncMono" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } - - def "test WithSpan annotated async method (Flux)"() { - setup: - def latch = new CountDownLatch(1) - def flux = ReactorTracedMethods.traceAsyncFlux(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - flux.blockLast() - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFlux" - operationName "ReactorTracedMethods.traceAsyncFlux" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } - - def "test WithSpan annotated async method (failing Flux)"() { - setup: - def latch = new CountDownLatch(1) - def expectedException = new IllegalStateException("Test exception") - def flux = ReactorTracedMethods.traceAsyncFailingFlux(latch, expectedException) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - flux.blockLast() - - then: - thrown(IllegalStateException) - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFailingFlux" - operationName "ReactorTracedMethods.traceAsyncFailingFlux" - errored true - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - errorTags(expectedException) - } - } - } - } - } - - def "test WithSpan annotated async method (cancelled Flux)"() { - setup: - def latch = new CountDownLatch(1) - def flux = ReactorTracedMethods.traceAsyncFlux(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - flux.subscribe(new ReactorTracedMethods.CancelSubscriber<>()) - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFlux" - operationName "ReactorTracedMethods.traceAsyncFlux" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorCoreTest.groovy b/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorCoreTest.groovy deleted file mode 100644 index 09823e83e9e..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/ReactorCoreTest.groovy +++ /dev/null @@ -1,598 +0,0 @@ -import datadog.trace.agent.test.InstrumentationSpecification -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.opentelemetry.api.GlobalOpenTelemetry -import io.opentelemetry.api.trace.Span -import io.opentracing.util.GlobalTracer -import org.reactivestreams.Publisher -import org.reactivestreams.Subscriber -import org.reactivestreams.Subscription -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.core.scheduler.Schedulers -import reactor.util.context.Context -import spock.lang.Shared - -import java.time.Duration -import java.util.concurrent.CompletableFuture - -import static datadog.trace.agent.test.utils.TraceUtils.basicSpan -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 - -class ReactorCoreTest extends InstrumentationSpecification { - - public static final String EXCEPTION_MESSAGE = "test exception" - - @Shared - def addOne = { i -> - addOneFunc(i) - } - - @Shared - def addTwo = { i -> - addTwoFunc(i) - } - - @Shared - def throwException = { - throw new RuntimeException(EXCEPTION_MESSAGE) - } - - @Override - void configurePreAgent() { - super.configurePreAgent() - injectSysConfig("dd.trace.otel.enabled", "true") - } - - def "Publisher '#name' test"() { - when: - def result = assemblePublisherUnderTrace(publisherSupplier) - - then: - result == expected - and: - assertTraces(1) { - sortSpansByStart() - trace(workSpans + 2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - - for (int i = 0; i < workSpans; i++) { - span { - resourceName "addOne" - operationName "addOne" - childOf(span(1)) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - // spotless:off - name | expected | workSpans | publisherSupplier - "basic mono" | 2 | 1 | { -> Mono.just(1).map(addOne) } - "two operations mono" | 4 | 2 | { -> Mono.just(2).map(addOne).map(addOne) } - "delayed mono" | 4 | 1 | { -> Mono.just(3).delayElement(Duration.ofMillis(100)).map(addOne) } - "delayed twice mono" | 6 | 2 | { -> - Mono.just(4).delayElement(Duration.ofMillis(100)).map(addOne).delayElement(Duration.ofMillis(100)).map(addOne) - } - "basic flux" | [6, 7] | 2 | { -> Flux.fromIterable([5, 6]).map(addOne) } - "two operations flux" | [8, 9] | 4 | { -> Flux.fromIterable([6, 7]).map(addOne).map(addOne) } - "delayed flux" | [8, 9] | 2 | { -> - Flux.fromIterable([7, 8]).delayElements(Duration.ofMillis(100)).map(addOne) - } - "delayed twice flux" | [10, 11] | 4 | { -> - Flux.fromIterable([8, 9]).delayElements(Duration.ofMillis(100)).map(addOne).delayElements(Duration.ofMillis(100)).map(addOne) - } - "mono from callable" | 12 | 2 | { -> Mono.fromCallable({ addOneFunc(10) }).map(addOne) } - // spotless:on - } - - def "Publisher error '#name' test"() { - when: - assemblePublisherUnderTrace(publisherSupplier) - - then: - def exception = thrown RuntimeException - exception.message == EXCEPTION_MESSAGE - and: - assertTraces(1) { - sortSpansByStart() - trace(2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - errored true - tags { - "$Tags.COMPONENT" "trace" - errorTags(RuntimeException, EXCEPTION_MESSAGE) - defaultTags() - } - } - - // It's important that we don't attach errors at the Reactor level so that we don't - // impact the spans on reactor integrations such as netty and lettuce, as reactor is - // more of a context propagation mechanism than something we would be tracking for - // errors this is ok. - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - } - } - - where: - name | publisherSupplier - "mono" | { -> Mono.error(new RuntimeException(EXCEPTION_MESSAGE)) } - "flux" | { -> Flux.error(new RuntimeException(EXCEPTION_MESSAGE)) } - } - - def "Publisher step '#name' test"() { - when: - assemblePublisherUnderTrace(publisherSupplier) - - then: - def exception = thrown RuntimeException - exception.message == EXCEPTION_MESSAGE - and: - assertTraces(1) { - sortSpansByStart() - trace(workSpans + 2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - errored true - tags { - "$Tags.COMPONENT" "trace" - errorTags(RuntimeException, EXCEPTION_MESSAGE) - defaultTags() - } - } - - // It's important that we don't attach errors at the Reactor level so that we don't - // impact the spans on reactor integrations such as netty and lettuce, as reactor is - // more of a context propagation mechanism than something we would be tracking for - // errors this is ok. - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - - for (int i = 0; i < workSpans; i++) { - span { - resourceName "addOne" - operationName "addOne" - childOf(span(1)) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - name | workSpans | publisherSupplier - "basic mono failure" | 1 | { -> Mono.just(1).map(addOne).map({ throwException() }) } - "basic flux failure" | 1 | { -> Flux.fromIterable([5, 6]).map(addOne).map({ throwException() }) } - } - - def "Publisher '#name' cancel"() { - when: - cancelUnderTrace(publisherSupplier) - - then: - assertTraces(1) { - trace(2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - } - } - - where: - name | publisherSupplier - "basic mono" | { -> Mono.just(1) } - "basic flux" | { -> Flux.fromIterable([5, 6]) } - } - - def "Publisher chain spans have the correct parent for '#name'"() { - when: - assemblePublisherUnderTrace(publisherSupplier) - - then: - assertTraces(1) { - trace(workSpans + 2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - - for (int i = 0; i < workSpans; i++) { - span { - resourceName "addOne" - operationName "addOne" - childOf(span(1)) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - name | workSpans | publisherSupplier - "basic mono" | 3 | { -> Mono.just(1).map(addOne).map(addOne).then(Mono.just(1).map(addOne)) } - "basic flux" | 5 | { -> Flux.fromIterable([5, 6]).map(addOne).map(addOne).then(Mono.just(1).map(addOne)) } - } - - def "Publisher chain spans have the correct parents from subscription time"() { - when: - def mono = Mono.just(42) - .map(addOne) - .map(addTwo) - - runUnderTrace("trace-parent") { - mono.block() - } - - then: - assertTraces(1) { - trace(3) { - sortSpansByStart() - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - } - span { - operationName "addOne" - childOf span(0) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - span { - operationName "addTwo" - childOf span(0) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - def "Publisher chain spans have the correct parents from subscription time '#name'"() { - when: - assemblePublisherUnderTrace { - // The "add one" operations in the publisher created here should be children of the publisher-parent - Publisher publisher = publisherSupplier() - - AgentSpan intermediate = startSpan("test", "intermediate") - AgentScope scope = activateSpan(intermediate) - try { - if (publisher instanceof Mono) { - return ((Mono) publisher).map(addTwo) - } else if (publisher instanceof Flux) { - return ((Flux) publisher).map(addTwo) - } - throw new IllegalStateException("Unknown publisher type") - } finally { - intermediate.finish() - scope.close() - } - } - - then: - assertTraces(1) { - trace(3 + 2 * workItems) { - sortSpansByStart() - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - basicSpan(it, "intermediate", span(1)) - - for (int i = 0; i < 2 * workItems; i = i + 2) { - span { - operationName "addOne" - childOf span(1) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - span { - operationName "addTwo" - childOf span(1) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - name | workItems | publisherSupplier - "basic mono" | 1 | { -> Mono.just(1).map(addOne) } - "basic flux" | 2 | { -> Flux.fromIterable([1, 2]).map(addOne) } - } - - def "Fluxes produce the right number of results on '#schedulerName' scheduler"() { - when: - List values = Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) - .parallel() - .runOn(scheduler) - .flatMap({ num -> Mono.just(num.toString() + " on " + Thread.currentThread().getName()) }) - .sequential() - .collectList() - .block() - - then: - values.size() == 4 - - where: - schedulerName | scheduler - "parallel" | Schedulers.parallel() - "elastic" | Schedulers.boundedElastic() - "single" | Schedulers.single() - "immediate" | Schedulers.immediate() - } - - def "subscribe-time context propagates across threads with #name"() { - // Guards that the thread-confined publisher hand-off (HandoffContext) does not break cross-thread - // propagation: the @Trace "addOne" spans run in map's onNext on scheduler threads and must still - // be children of the subscribe-time parent. - when: - runUnderTrace("parent") { - pipeline.call().collectList().block() - } - - then: - assertTraces(1) { - trace(3) { - sortSpansByStart() - span { - operationName "parent" - parent() - } - span { - operationName "addOne" - childOf span(0) - } - span { - operationName "addOne" - childOf span(0) - } - } - } - - where: - name | pipeline - "publishOn" | { - Flux.just(1, 2).publishOn(Schedulers.parallel()).map(addOne) - } - "subscribeOn" | { - Flux.just(1, 2).subscribeOn(Schedulers.single()).map(addOne) - } - "subscribeOn+publishOn" | { - Flux.just(1, 2) - .subscribeOn(Schedulers.single()) - .publishOn(Schedulers.parallel()) - .map(addOne) - } - } - - def "Context propagation through reactor context with span #spanType"() { - when: - runUnderTrace("parent", { - Mono.defer { - def span = buildSpan() - Mono.just(0) - .contextWrite(Context.of("dd.span", span)) - .doFinally { - ignored -> finishSpan(span) - } - } - .map(this::addOneFunc) - .block() - }) - then: - assertTraces(1, { - trace(3) { - basicSpan(it, "parent") - basicSpan(it, "contextual", span(0), null, ['span.kind': { - true - }]) // otel spans sets span.kind - span { - operationName "addOne" - childOfPrevious() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - }) - where: - spanType | buildSpan | finishSpan - "datadog" | { - TEST_TRACER.buildSpan("reactor-core", "contextual").start() - } | { - AgentSpan span -> span.finish() - } - "opentracing" | { - GlobalTracer.get().buildSpan("contextual").start() - } | { - io.opentracing.Span span -> span.finish() - } - "otel" | { - GlobalOpenTelemetry.get().getTracer("").spanBuilder("contextual").setAttribute("operation.name", "contextual").startSpan() - } | { - Span span -> span.end() - } - } - - def "Bad values for dd.span are tolerated"() { - when: - runUnderTrace("parent", { - Mono.just(1) - .contextWrite(Context.of("dd.span", "Hello world")) - .map(this::addOneFunc) - .block() - }) - then: - assertTraces(1, { - trace(2) { - basicSpan(it, "parent") - span { - operationName "addOne" - childOfPrevious() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - }) - } - - def "test currentContext() calls on inner operator is not throwing a NPE on the advice"() { - when: - def mono = Flux.range(1, 100).windowUntil { it % 10 == 0 }.count() - then: - // we are not interested into asserting a trace structure but only that the instrumentation error count is 0 - assert mono.block() == 10 - } - - def "span in the context has to be activated when the publisher subscribes"() { - when: - // the mono is subscribed (block) when first is active. - // However we expect that the span third will have second as parent and not first - // because we set the parent explicitly in the reactor context (dd.span key) - def result = runUnderTrace("first", { - runUnderTrace("second", { - def mono = Mono.defer { - Mono.fromCompletionStage(CompletableFuture.supplyAsync { - runUnderTrace("third", { - "hello world" - }) - }) - }.contextWrite(Context.of("dd.span", TEST_TRACER.activeSpan())) - mono - }) - .block() - }) - then: - assert result == "hello world" - assertTraces(1, { - trace(3, true) { - basicSpan(it, "first") - basicSpan(it, "second", span(0)) - basicSpan(it, "third", span(1)) - } - }) - } - - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - def assemblePublisherUnderTrace(def publisherSupplier) { - def span = startSpan("test", "publisher-parent") - // After this activation, the "add two" operations below should be children of this span - def scope = activateSpan(span) - - Publisher publisher = publisherSupplier() - try { - // Read all data from publisher - if (publisher instanceof Mono) { - return publisher.block() - } else if (publisher instanceof Flux) { - return publisher.toStream().toArray({ size -> new Integer[size] }) - } - - throw new RuntimeException("Unknown publisher: " + publisher) - } finally { - span.finish() - scope.close() - } - } - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - def cancelUnderTrace(def publisherSupplier) { - final AgentSpan span = startSpan("test", "publisher-parent") - AgentScope scope = activateSpan(span) - - def publisher = publisherSupplier() - publisher.subscribe(new Subscriber() { - void onSubscribe(Subscription subscription) { - subscription.cancel() - } - - void onNext(Integer t) { - } - - void onError(Throwable error) { - } - - void onComplete() { - } - }) - - scope.close() - span.finish() - } - - @Trace(operationName = "addOne", resourceName = "addOne") - def static addOneFunc(int i) { - return i + 1 - } - - @Trace(operationName = "addTwo", resourceName = "addTwo") - def static addTwoFunc(int i) { - return i + 2 - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/SubscriptionTest.groovy b/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/SubscriptionTest.groovy deleted file mode 100644 index 40992938fbd..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/groovy/SubscriptionTest.groovy +++ /dev/null @@ -1,126 +0,0 @@ -import static datadog.trace.agent.test.utils.TraceUtils.basicSpan -import static datadog.trace.agent.test.utils.TraceUtils.runnableUnderTrace - -import datadog.trace.agent.test.InstrumentationSpecification -import datadog.trace.bootstrap.instrumentation.api.AgentTracer -import reactor.core.publisher.Flux -import reactor.core.publisher.Sinks - -import java.util.concurrent.CountDownLatch - -class SubscriptionTest extends InstrumentationSpecification { - - def "subscription test with Sinks.Many sink #sink.class and #consumers consumers"() { - when: - def connection = ((Sinks.Many) sink) - CountDownLatch published = new CountDownLatch(consumers) - // need to wait for subscriber when using a non buffered processor - CountDownLatch subscribed = new CountDownLatch(consumers) - for (int i = 0; i < consumers; i++) { - def t = new Thread({ - runnableUnderTrace("parent", { - connection.asFlux().subscribe { - it.query() - published.countDown() - } - subscribed.countDown() - published.await() - }) - }) - t.start() - } - subscribed.await() - connection.tryEmitNext(new Connection()).orThrow() - published.await() - then: - assertTraces(consumers) { - for (int i = 0; i< consumers; i++) { - trace(2) { - basicSpan(it, "parent") - basicSpan(it, "Connection.query", span(0)) - } - } - } - where: - sink | consumers - Sinks.many().unicast().onBackpressureBuffer() | 1 - Sinks.many().multicast().onBackpressureBuffer() | 1 - Sinks.many().multicast().directAllOrNothing() | 1 - Sinks.many().multicast().directBestEffort() | 1 - Sinks.many().replay().all() | 1 - Sinks.many().replay().latest() | 1 - Sinks.many().multicast().onBackpressureBuffer() | 3 - Sinks.many().multicast().directAllOrNothing() | 3 - Sinks.many().multicast().directBestEffort() | 3 - Sinks.many().replay().all() | 3 - Sinks.many().replay().latest() | 3 - } - - def "subscription test with Sinks.One sink"() { - when: - def connection = Sinks. one() - CountDownLatch published = new CountDownLatch(1) - // need to wait for subscriber when using a non buffered processor - CountDownLatch subscribed = new CountDownLatch(1) - def t = new Thread({ - runnableUnderTrace("parent", { - connection.asMono().subscribe { - it.query() - published.countDown() - } - subscribed.countDown() - published.await() - }) - }) - t.start() - subscribed.await() - connection.tryEmitValue(new Connection()).orThrow() - published.await() - t.join() - then: - assertTraces(1) { - trace(2) { - basicSpan(it, "parent") - basicSpan(it, "Connection.query", span(0)) - } - } - } - - def "test broadcasting flux"() { - when: - def connection = Flux. just(new Connection()).publish() - - CountDownLatch published = new CountDownLatch(1) - // need to wait for subscriber when using a non buffered processor - CountDownLatch subscribed = new CountDownLatch(1) - def t = new Thread({ - runnableUnderTrace("parent", { - connection.subscribe { - it.query() - published.countDown() - } - subscribed.countDown() - published.await() - }) - }) - t.start() - subscribed.await() - connection.connect() - published.await() - then: - assertTraces(1) { - trace(2) { - basicSpan(it, "parent") - basicSpan(it, "Connection.query", span(0)) - } - } - } - - static class Connection { - static int query() { - def span = AgentTracer.startSpan("test", "Connection.query") - span.finish() - return new Random().nextInt() - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/java/annotatedsample/ReactorTracedMethods.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/java/annotatedsample/ReactorTracedMethods.java deleted file mode 100644 index a3450a31e20..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/latestDepTest/java/annotatedsample/ReactorTracedMethods.java +++ /dev/null @@ -1,75 +0,0 @@ -package annotatedsample; - -import static java.util.concurrent.TimeUnit.SECONDS; - -import io.opentelemetry.instrumentation.annotations.WithSpan; -import java.util.concurrent.CountDownLatch; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -public class ReactorTracedMethods { - @WithSpan - public static Mono traceAsyncMono(CountDownLatch latch) { - return Mono.fromCallable( - () -> { - await(latch); - return "hello"; - }); - } - - @WithSpan - public static Mono traceAsyncFailingMono(CountDownLatch latch, Exception exception) { - return Mono.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Flux traceAsyncFlux(CountDownLatch latch) { - return Flux.create( - emitter -> { - await(latch); - emitter.next("hello"); - emitter.complete(); - }); - } - - @WithSpan - public static Flux traceAsyncFailingFlux(CountDownLatch latch, Exception exception) { - return Flux.create( - emitter -> { - await(latch); - emitter.error(exception); - }); - } - - private static void await(CountDownLatch latch) { - try { - if (!latch.await(5, SECONDS)) { - throw new IllegalStateException("Latch still locked"); - } - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - - public static class CancelSubscriber implements Subscriber { - @Override - public void onSubscribe(Subscription s) { - s.cancel(); - } - - @Override - public void onNext(T t) {} - - @Override - public void onError(Throwable t) {} - - @Override - public void onComplete() {} - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/BlockingPublisherInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/BlockingPublisherInstrumentation.java deleted file mode 100644 index 6e0bb195f9c..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/BlockingPublisherInstrumentation.java +++ /dev/null @@ -1,55 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import static datadog.trace.agent.tooling.bytebuddy.matcher.HierarchyMatchers.hasSuperType; -import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.nameStartsWith; -import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.namedOneOf; -import static net.bytebuddy.matcher.ElementMatchers.isMethod; - -import datadog.context.ContextScope; -import datadog.trace.agent.tooling.Instrumenter; -import datadog.trace.bootstrap.InstrumentationContext; -import datadog.trace.bootstrap.instrumentation.reactivestreams.HandoffContext; -import net.bytebuddy.asm.Advice; -import net.bytebuddy.description.type.TypeDescription; -import net.bytebuddy.matcher.ElementMatcher; -import org.reactivestreams.Publisher; - -/** - * This instrumentation is responsible for capturing the right state when Mono or Flux block* - * methods are called. This because the mechanism they handle this differs a bit of the standard - * {@link Publisher#subscribe} - */ -public class BlockingPublisherInstrumentation - implements Instrumenter.ForTypeHierarchy, Instrumenter.HasMethodAdvice { - - @Override - public String hierarchyMarkerType() { - return "reactor.core.publisher.Mono"; - } - - @Override - public ElementMatcher hierarchyMatcher() { - return hasSuperType(namedOneOf("reactor.core.publisher.Mono", "reactor.core.publisher.Flux")); - } - - @Override - public void methodAdvice(MethodTransformer transformer) { - transformer.applyAdvice( - isMethod().and(nameStartsWith("block")), getClass().getName() + "$BlockingAdvice"); - } - - public static class BlockingAdvice { - @Advice.OnMethodEnter(suppress = Throwable.class) - public static ContextScope before(@Advice.This final Publisher self) { - return ReactorContextBridge.activateForBlocking( - self, InstrumentationContext.get(Publisher.class, HandoffContext.class)); - } - - @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class) - public static void after(@Advice.Enter final ContextScope scope) { - if (scope != null) { - scope.close(); - } - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ContextWritingSubscriberInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ContextWritingSubscriberInstrumentation.java deleted file mode 100644 index 87f407fae94..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ContextWritingSubscriberInstrumentation.java +++ /dev/null @@ -1,71 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.namedOneOf; -import static net.bytebuddy.matcher.ElementMatchers.isConstructor; - -import datadog.context.Context; -import datadog.context.ContextScope; -import datadog.trace.agent.tooling.Instrumenter; -import datadog.trace.bootstrap.InstrumentationContext; -import net.bytebuddy.asm.Advice; -import org.reactivestreams.Subscriber; -import reactor.core.CoreSubscriber; - -/** - * Tailored instrumentation for Reactor's context-writing subscribers (the inner subscribers created - * by {@code contextWrite}/{@code subscriberContext} operators). Matching them by exact type removes - * the per-signal {@code instanceof} chain that the generic {@code CoreSubscriberInstrumentation} - * used to run on every {@link CoreSubscriber}. - * - *

The explicit Datadog {@link Context} carried by these subscribers is fixed at construction, so - * it is read once (constructor advice) and stored; signal advice then only does a context-store - * lookup. - */ -public class ContextWritingSubscriberInstrumentation - implements Instrumenter.ForKnownTypes, Instrumenter.HasMethodAdvice { - - @Override - public String[] knownMatchingTypes() { - return new String[] { - "reactor.core.publisher.FluxContextWrite$ContextWriteSubscriber", - "reactor.core.publisher.FluxContextStart$ContextStartSubscriber", - "reactor.core.publisher.FluxContextWriteRestoringThreadLocals" - + "$ContextWriteRestoringThreadLocalsSubscriber", - "reactor.core.publisher.FluxContextWriteRestoringThreadLocalsFuseable" - + "$FuseableContextWriteRestoringThreadLocalsSubscriber", - "reactor.core.publisher.MonoContextWriteRestoringThreadLocals" - + "$ContextWriteRestoringThreadLocalsSubscriber", - }; - } - - @Override - public void methodAdvice(MethodTransformer transformer) { - transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureContextAdvice"); - transformer.applyAdvice( - namedOneOf("onNext", "onComplete", "onError"), - getClass().getName() + "$ActivateContextAdvice"); - } - - public static class CaptureContextAdvice { - @Advice.OnMethodExit(suppress = Throwable.class) - public static void onConstructed(@Advice.This final CoreSubscriber self) { - ReactorContextBridge.captureSubscriberContext( - self, InstrumentationContext.get(Subscriber.class, Context.class)); - } - } - - public static class ActivateContextAdvice { - @Advice.OnMethodEnter(suppress = Throwable.class) - public static ContextScope before(@Advice.This final CoreSubscriber self) { - return ReactorContextBridge.activateStoredContext( - self, InstrumentationContext.get(Subscriber.class, Context.class)); - } - - @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class) - public static void after(@Advice.Enter final ContextScope scope) { - if (scope != null) { - scope.close(); - } - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/CorePublisherInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/CorePublisherInstrumentation.java deleted file mode 100644 index 11e025b0978..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/CorePublisherInstrumentation.java +++ /dev/null @@ -1,73 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import static datadog.trace.agent.tooling.bytebuddy.matcher.HierarchyMatchers.hasSuperType; -import static datadog.trace.agent.tooling.bytebuddy.matcher.HierarchyMatchers.implementsInterface; -import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; -import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.namedOneOf; -import static net.bytebuddy.matcher.ElementMatchers.isStatic; -import static net.bytebuddy.matcher.ElementMatchers.not; -import static net.bytebuddy.matcher.ElementMatchers.takesArgument; -import static net.bytebuddy.matcher.ElementMatchers.takesArguments; - -import datadog.context.Context; -import datadog.context.ContextScope; -import datadog.trace.agent.tooling.Instrumenter; -import datadog.trace.bootstrap.InstrumentationContext; -import datadog.trace.bootstrap.instrumentation.reactivestreams.HandoffContext; -import net.bytebuddy.asm.Advice; -import net.bytebuddy.description.type.TypeDescription; -import net.bytebuddy.matcher.ElementMatcher; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import reactor.core.CoreSubscriber; - -public class CorePublisherInstrumentation - implements Instrumenter.ForTypeHierarchy, Instrumenter.HasMethodAdvice { - - @Override - public String hierarchyMarkerType() { - return "reactor.core.CoreSubscriber"; - } - - @Override - public ElementMatcher hierarchyMatcher() { - return implementsInterface(named("reactor.core.CorePublisher")) // from 3.1.7 - .or( - hasSuperType( - namedOneOf( - "reactor.core.publisher.Mono", "reactor.core.publisher.Flux"))); // < 3.1.7 - } - - @Override - public void methodAdvice(MethodTransformer transformer) { - transformer.applyAdvice( - named("subscribe") - .and(not(isStatic())) - .and(takesArguments(1)) - .and(takesArgument(0, named("reactor.core.CoreSubscriber"))), - getClass().getName() + "$PropagateContextSpanOnSubscribe"); - } - - public static class PropagateContextSpanOnSubscribe { - @Advice.OnMethodEnter(suppress = Throwable.class) - public static ContextScope before( - @Advice.This final Publisher self, - @Advice.Argument(0) final CoreSubscriber subscriber) { - // Hands the explicit context recorded for a context-writing subscriber to the publisher store - // (for the reactive-streams hand-off) and attaches it. The subscriber wrapping for - // context-reading operators lives in ContextReadingPublisherInstrumentation. - return ReactorContextBridge.captureOnSubscribe( - self, - subscriber, - InstrumentationContext.get(Publisher.class, HandoffContext.class), - InstrumentationContext.get(Subscriber.class, Context.class)); - } - - @Advice.OnMethodExit(suppress = Throwable.class, onThrowable = Throwable.class) - public static void after(@Advice.Enter final ContextScope scope) { - if (scope != null) { - scope.close(); - } - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/OptimizableOperatorInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/OptimizableOperatorInstrumentation.java deleted file mode 100644 index 73c963dd898..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/OptimizableOperatorInstrumentation.java +++ /dev/null @@ -1,62 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import static datadog.trace.agent.tooling.bytebuddy.matcher.HierarchyMatchers.hasInterface; -import static datadog.trace.agent.tooling.bytebuddy.matcher.HierarchyMatchers.implementsInterface; -import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; -import static net.bytebuddy.matcher.ElementMatchers.isMethod; -import static net.bytebuddy.matcher.ElementMatchers.returns; -import static net.bytebuddy.matcher.ElementMatchers.takesArguments; - -import datadog.context.Context; -import datadog.trace.agent.tooling.Instrumenter; -import datadog.trace.bootstrap.InstrumentationContext; -import datadog.trace.bootstrap.instrumentation.reactivestreams.HandoffContext; -import net.bytebuddy.asm.Advice; -import net.bytebuddy.description.type.TypeDescription; -import net.bytebuddy.matcher.ElementMatcher; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; - -/** - * This instrumentation is responsible for transferring the {@link Context} when subscription - * optimization are made. In particular reactor's OptimizableOperators can do subscription via a - * loop call instead of recursion. - */ -public class OptimizableOperatorInstrumentation - implements Instrumenter.ForTypeHierarchy, Instrumenter.HasMethodAdvice { - - @Override - public String hierarchyMarkerType() { - return "reactor.core.publisher.OptimizableOperator"; - } - - @Override - public ElementMatcher hierarchyMatcher() { - return implementsInterface(named(hierarchyMarkerType())); - } - - @Override - public void methodAdvice(MethodTransformer transformer) { - transformer.applyAdvice( - isMethod() - .and(named("subscribeOrReturn")) - .and(takesArguments(1)) - .and(returns(hasInterface(named("org.reactivestreams.Subscriber")))), - getClass().getName() + "$PublisherSubscribeAdvice"); - } - - public static class PublisherSubscribeAdvice { - @Advice.OnMethodExit(suppress = Throwable.class) - public static void onSubscribe( - @Advice.This final Publisher self, - @Advice.Argument(0) final Subscriber arg, - @Advice.Return final Subscriber s) { - ReactorContextBridge.transferToOptimizedSubscriber( - self, - arg, - s, - InstrumentationContext.get(Publisher.class, HandoffContext.class), - InstrumentationContext.get(Subscriber.class, Context.class)); - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorContextBridge.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorContextBridge.java deleted file mode 100644 index df2f9a91a0c..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorContextBridge.java +++ /dev/null @@ -1,139 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import datadog.context.Context; -import datadog.context.ContextScope; -import datadog.trace.bootstrap.ContextStore; -import datadog.trace.bootstrap.instrumentation.api.WithAgentSpan; -import datadog.trace.bootstrap.instrumentation.reactivestreams.HandoffContext; -import org.reactivestreams.Publisher; -import org.reactivestreams.Subscriber; -import reactor.core.CoreSubscriber; - -/** - * Helper shared by the reactor-core instrumentations. It reads the span a user placed in the - * Reactor context under the {@code dd.span} key (adapting it to a {@link Context}) and drives the - * context-store-based propagation: capture on subscribe, restore on signal/blocking, and transfer - * to optimized subscribers. - */ -public final class ReactorContextBridge { - - private static final String DD_SPAN_KEY = "dd.span"; - - private ReactorContextBridge() {} - - /** - * Records the {@link Context} derived from the {@code dd.span} span a context-writing subscriber - * carries, into the subscriber store, once, when the subscriber is constructed. The caller - * ({@code ContextWritingSubscriberInstrumentation}) only matches context-writing subscribers, so - * no runtime type check is needed. - */ - public static void captureSubscriberContext( - final CoreSubscriber subscriber, - final ContextStore subscriberContexts) { - final Context context = explicitContextFromSubscriber(subscriber); - if (context != null) { - subscriberContexts.put(subscriber, context); - } - } - - /** - * Attaches the context recorded for {@code subscriber} by {@link #captureSubscriberContext}. A - * plain store lookup — no {@code instanceof}, no {@code currentContext()} call — on the signal - * hot path. - */ - public static ContextScope activateStoredContext( - final Subscriber subscriber, final ContextStore subscriberContexts) { - return attachIfRequired(subscriberContexts.get(subscriber), Context.current()); - } - - /** - * On subscribe, hands the explicit context recorded for {@code subscriber} (a context-writing - * subscriber) to the publisher store so the reactive-streams layer can propagate it, and attaches - * it. The deposit is {@linkplain HandoffContext#threadConfined thread-confined} since the - * subscribed publisher may be a concurrently-subscribed shared sink. - */ - public static ContextScope captureOnSubscribe( - final Publisher publisher, - final Subscriber subscriber, - final ContextStore publisherContexts, - final ContextStore subscriberContexts) { - final Context context = subscriberContexts.get(subscriber); - if (context == null) { - return null; - } - - publisherContexts.put(publisher, HandoffContext.threadConfined(context)); - return attachIfRequired(context, Context.current()); - } - - public static ContextScope activateForBlocking( - final Publisher publisher, - final ContextStore publisherContexts) { - final HandoffContext handoff = publisherContexts.get(publisher); - return attachIfRequired( - handoff == null ? null : handoff.contextForCurrentThread(), Context.current()); - } - - public static void transferToOptimizedSubscriber( - final Publisher publisher, - final Subscriber source, - final Subscriber target, - final ContextStore publisherContexts, - final ContextStore subscriberContexts) { - if (source == null || target == null) { - return; - } - - final HandoffContext handoff = publisherContexts.get(publisher); - Context context = handoff == null ? null : handoff.contextForCurrentThread(); - if (context == null) { - context = subscriberContexts.get(source); - } - if (context != null) { - subscriberContexts.putIfAbsent(target, context); - } - } - - private static Context explicitContextFromSubscriber(final CoreSubscriber subscriber) { - final reactor.util.context.Context reactorContext = currentContext(subscriber); - if (reactorContext == null || !hasKey(reactorContext, DD_SPAN_KEY)) { - return null; - } - final Object maybeSpan = get(reactorContext, DD_SPAN_KEY); - return maybeSpan instanceof WithAgentSpan ? ((WithAgentSpan) maybeSpan).asAgentSpan() : null; - } - - private static reactor.util.context.Context currentContext(final CoreSubscriber subscriber) { - if (subscriber == null) { - return null; - } - try { - return subscriber.currentContext(); - } catch (Throwable ignored) { - return null; - } - } - - private static ContextScope attachIfRequired(final Context context, final Context activeContext) { - if (context == null || context == activeContext || context == Context.root()) { - return null; - } - return context.attach(); - } - - private static boolean hasKey(final reactor.util.context.Context context, final Object key) { - try { - return context.hasKey(key); - } catch (Throwable ignored) { - return false; - } - } - - private static Object get(final reactor.util.context.Context context, final Object key) { - try { - return context.get(key); - } catch (Throwable ignored) { - return null; - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorCoreModule.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorCoreModule.java deleted file mode 100644 index 14bd109c458..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorCoreModule.java +++ /dev/null @@ -1,62 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import static java.util.Arrays.asList; - -import com.google.auto.service.AutoService; -import datadog.context.Context; -import datadog.trace.agent.tooling.ExcludeFilterProvider; -import datadog.trace.agent.tooling.Instrumenter; -import datadog.trace.agent.tooling.InstrumenterModule; -import datadog.trace.bootstrap.instrumentation.java.concurrent.ExcludeFilter; -import datadog.trace.bootstrap.instrumentation.reactivestreams.HandoffContext; -import java.util.Arrays; -import java.util.Collection; -import java.util.Collections; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -@AutoService(InstrumenterModule.class) -public final class ReactorCoreModule extends InstrumenterModule.ContextTracking - implements ExcludeFilterProvider { - public ReactorCoreModule() { - super("reactor-core"); - } - - @Override - public String[] helperClassNames() { - return new String[] { - packageName + ".ReactorAsyncResultExtension", packageName + ".ReactorContextBridge", - }; - } - - @Override - public Map contextStore() { - final Map store = new HashMap<>(); - store.put("org.reactivestreams.Subscriber", Context.class.getName()); - store.put("org.reactivestreams.Publisher", HandoffContext.class.getName()); - return store; - } - - @Override - public Map> excludedClasses() { - return Collections.singletonMap( - ExcludeFilter.ExcludeType.RUNNABLE, - Arrays.asList( - "reactor.core.publisher.EventLoopProcessor", - "reactor.core.publisher.EventLoopProcessor$RequestTask", - "reactor.core.publisher.TopicProcessor$TopicInner", - "reactor.core.publisher.TopicProcessor$TopicInner$1", - "reactor.core.publisher.WorkQueueProcessor$WorkQueueInner", - "reactor.core.publisher.WorkQueueProcessor$WorkQueueInner$1")); - } - - @Override - public List typeInstrumentations() { - return asList( - new BlockingPublisherInstrumentation(), - new CorePublisherInstrumentation(), - new ContextWritingSubscriberInstrumentation(), - new OptimizableOperatorInstrumentation()); - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java new file mode 100644 index 00000000000..904ddaefac8 --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java @@ -0,0 +1,70 @@ +package datadog.trace.instrumentation.reactorcore; + +import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; +import static datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge.currentContext; +import static datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge.rootContext; +import static net.bytebuddy.matcher.ElementMatchers.isConstructor; +import static net.bytebuddy.matcher.ElementMatchers.isMethod; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import datadog.context.Context; +import datadog.context.ContextScope; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.bootstrap.InstrumentationContext; +import net.bytebuddy.asm.Advice; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.Flux; + +public final class FluxInstrumentation + implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { + @Override + public String instrumentedType() { + return "reactor.core.publisher.Flux"; + } + + @Override + public void methodAdvice(MethodTransformer transformer) { + transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice"); + transformer.applyAdvice( + isMethod() + .and(named("subscribe")) + .and(takesArguments(1)) + .and(takesArgument(0, named("reactor.core.CoreSubscriber"))), + getClass().getName() + "$PropagateParentSpanAdvice"); + } + + public static class CaptureParentSpanAdvice { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onConstruct(@Advice.This final Flux flux) { + Context parentContext = currentContext(); + if (parentContext != rootContext()) { + InstrumentationContext.get(Flux.class, Context.class).put(flux, parentContext); + } + } + } + + public static class PropagateParentSpanAdvice { + @Advice.OnMethodEnter(suppress = Throwable.class) + public static ContextScope onSubscribe( + @Advice.This final Flux flux, + @Advice.Argument(value = 0, readOnly = false) CoreSubscriber subscriber) { + if (subscriber != null) { + Context parentContext = + InstrumentationContext.get(Flux.class, Context.class).get(flux); + if (parentContext != null) { + subscriber = new TracingCoreSubscriber<>(subscriber, parentContext); + return parentContext.attach(); + } + } + return null; + } + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) + public static void closeScope(@Advice.Enter final ContextScope scope) { + if (scope != null) { + scope.close(); + } + } + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java new file mode 100644 index 00000000000..6ecc9d4cdd5 --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java @@ -0,0 +1,72 @@ +package datadog.trace.instrumentation.reactorcore; + +import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; +import static datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge.currentContext; +import static datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge.rootContext; +import static net.bytebuddy.matcher.ElementMatchers.isConstructor; +import static net.bytebuddy.matcher.ElementMatchers.isMethod; +import static net.bytebuddy.matcher.ElementMatchers.takesArgument; +import static net.bytebuddy.matcher.ElementMatchers.takesArguments; + +import datadog.context.Context; +import datadog.context.ContextScope; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.bootstrap.InstrumentationContext; +import net.bytebuddy.asm.Advice; +import reactor.core.CoreSubscriber; +import reactor.core.publisher.Mono; + +public final class MonoInstrumentation + implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { + @Override + public String instrumentedType() { + return "reactor.core.publisher.Mono"; + } + + @Override + public void methodAdvice(MethodTransformer transformer) { + transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice"); + transformer.applyAdvice( + isMethod() + .and(named("subscribe")) + .and(takesArguments(1)) + .and(takesArgument(0, named("reactor.core.CoreSubscriber"))), + getClass().getName() + "$PropagateParentSpanAdvice"); + } + + public static class CaptureParentSpanAdvice { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void onConstruct(@Advice.This final Mono mono) { + Context parentContext = currentContext(); + if (parentContext != rootContext()) { + InstrumentationContext.get(Mono.class, Context.class).put(mono, parentContext); + } + } + } + + public static class PropagateParentSpanAdvice { + @Advice.OnMethodEnter(suppress = Throwable.class) + public static ContextScope onSubscribe( + @Advice.This final Mono mono, + @Advice.Argument(value = 0, readOnly = false) CoreSubscriber subscriber) { + if (subscriber != null) { + Context parentContext = + InstrumentationContext.get(Mono.class, Context.class).get(mono); + if (parentContext != null) { + // wrap the subscriber so spans from its events treat the captured span as their parent + subscriber = new TracingCoreSubscriber<>(subscriber, parentContext); + // attach the context here in case additional subscribers are created during subscribe + return parentContext.attach(); + } + } + return null; + } + + @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) + public static void closeScope(@Advice.Enter final ContextScope scope) { + if (scope != null) { + scope.close(); + } + } + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorAsyncResultExtension.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorAsyncResultExtension.java new file mode 100644 index 00000000000..d1b3d162a8f --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorAsyncResultExtension.java @@ -0,0 +1,49 @@ +package datadog.trace.instrumentation.reactorcore; + +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.EagerHelper; +import datadog.trace.bootstrap.instrumentation.java.concurrent.AsyncResultExtension; +import datadog.trace.bootstrap.instrumentation.java.concurrent.AsyncResultExtensions; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +public class ReactorAsyncResultExtension implements AsyncResultExtension, EagerHelper { + static { + AsyncResultExtensions.register(new ReactorAsyncResultExtension()); + } + + /** + * Register the extension as an {@link AsyncResultExtension} using static class initialization. + *
+ * It uses an empty static method call to ensure the class loading and the one-time-only static + * class initialization. This will ensure this extension will only be registered once under {@link + * AsyncResultExtensions}. + */ + public static void init() {} + + @Override + public boolean supports(Class result) { + return Mono.class.isAssignableFrom(result) || Flux.class.isAssignableFrom(result); + } + + @Override + public Object apply(Object result, AgentSpan span) { + if (result instanceof Mono) { + return ((Mono) result) + .doOnError(throwable -> onError(span, throwable)) + .doOnCancel(span::finish) + .doOnSuccess(ignored -> span.finish()); + } else if (result instanceof Flux) { + return ((Flux) result) + .doOnComplete(span::finish) + .doOnError(throwable -> onError(span, throwable)) + .doOnCancel(span::finish); + } + return null; + } + + private static void onError(AgentSpan span, Throwable throwable) { + span.addThrowable(throwable); + span.finish(); + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java new file mode 100644 index 00000000000..499f47b26a0 --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java @@ -0,0 +1,39 @@ +package datadog.trace.instrumentation.reactorcore; + +import static java.util.Arrays.asList; + +import com.google.auto.service.AutoService; +import datadog.context.Context; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.agent.tooling.InstrumenterModule; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +@AutoService(InstrumenterModule.class) +public final class ReactorCoreModule extends InstrumenterModule.ContextTracking { + public ReactorCoreModule() { + super("reactor-core"); + } + + @Override + public String[] helperClassNames() { + return new String[] { + packageName + ".TracingCoreSubscriber", + packageName + ".ReactorAsyncResultExtension", + }; + } + + @Override + public Map contextStore() { + final Map store = new HashMap<>(); + store.put("reactor.core.publisher.Mono", Context.class.getName()); + store.put("reactor.core.publisher.Flux", Context.class.getName()); + return store; + } + + @Override + public List typeInstrumentations() { + return asList(new MonoInstrumentation(), new FluxInstrumentation()); + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/TracingCoreSubscriber.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/TracingCoreSubscriber.java new file mode 100644 index 00000000000..e3c9fb75727 --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/TracingCoreSubscriber.java @@ -0,0 +1,43 @@ +package datadog.trace.instrumentation.reactorcore; + +import datadog.context.Context; +import datadog.context.ContextScope; +import org.reactivestreams.Subscription; +import reactor.core.CoreSubscriber; + +/** Wrapper that makes sure spans from subscriber events treat the captured span as their parent. */ +public final class TracingCoreSubscriber implements CoreSubscriber { + private final CoreSubscriber delegate; + private final Context parentContext; + + public TracingCoreSubscriber(final CoreSubscriber delegate, final Context parentContext) { + this.delegate = delegate; + this.parentContext = parentContext; + } + + @Override + public void onSubscribe(final Subscription s) { + delegate.onSubscribe(s); + } + + @Override + public void onNext(final T value) { + try (final ContextScope scope = parentContext.attach()) { + delegate.onNext(value); + } + } + + @Override + public void onError(final Throwable t) { + try (final ContextScope scope = parentContext.attach()) { + delegate.onError(t); + } + } + + @Override + public void onComplete() { + try (final ContextScope scope = parentContext.attach()) { + delegate.onComplete(); + } + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorAsyncResultExtensionTest.groovy b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorAsyncResultExtensionTest.groovy deleted file mode 100644 index 5765cf21319..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorAsyncResultExtensionTest.groovy +++ /dev/null @@ -1,205 +0,0 @@ -import annotatedsample.ReactorTracedMethods -import com.google.common.util.concurrent.ListeningExecutorService -import com.google.common.util.concurrent.MoreExecutors -import datadog.trace.agent.test.InstrumentationSpecification -import datadog.trace.bootstrap.instrumentation.api.Tags -import spock.lang.Shared - -import java.util.concurrent.CountDownLatch -import java.util.concurrent.Executors - -class ReactorAsyncResultExtensionTest extends InstrumentationSpecification { - @Override - void configurePreAgent() { - super.configurePreAgent() - injectSysConfig("dd.trace.otel.enabled", "true") - injectSysConfig("dd.integration.opentelemetry-annotations-1.20.enabled", "true") - } - - @Shared - ListeningExecutorService executor - - def setupSpec() { - this.executor = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor()) - } - - def cleanupSpec() { - this.executor.shutdownNow() - } - - def "test WithSpan annotated async method (Mono)"() { - setup: - def latch = new CountDownLatch(1) - def mono = ReactorTracedMethods.traceAsyncMono(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - mono.block() - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncMono" - operationName "ReactorTracedMethods.traceAsyncMono" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } - - def "test WithSpan annotated async method (failing Mono)"() { - setup: - def latch = new CountDownLatch(1) - def expectedException = new IllegalStateException("Test exception") - def mono = ReactorTracedMethods.traceAsyncFailingMono(latch, expectedException) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - mono.block() - - then: - thrown(IllegalStateException) - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFailingMono" - operationName "ReactorTracedMethods.traceAsyncFailingMono" - errored true - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - errorTags(expectedException) - } - } - } - } - } - - def "test WithSpan annotated async method (cancelled Mono)"() { - setup: - def latch = new CountDownLatch(1) - def mono = ReactorTracedMethods.traceAsyncMono(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - mono.subscribe(new ReactorTracedMethods.CancelSubscriber<>()) - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncMono" - operationName "ReactorTracedMethods.traceAsyncMono" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } - - def "test WithSpan annotated async method (Flux)"() { - setup: - def latch = new CountDownLatch(1) - def flux = ReactorTracedMethods.traceAsyncFlux(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - flux.blockLast() - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFlux" - operationName "ReactorTracedMethods.traceAsyncFlux" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } - - def "test WithSpan annotated async method (failing Flux)"() { - setup: - def latch = new CountDownLatch(1) - def expectedException = new IllegalStateException("Test exception") - def flux = ReactorTracedMethods.traceAsyncFailingFlux(latch, expectedException) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - flux.blockLast() - - then: - thrown(IllegalStateException) - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFailingFlux" - operationName "ReactorTracedMethods.traceAsyncFailingFlux" - errored true - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - errorTags(expectedException) - } - } - } - } - } - - def "test WithSpan annotated async method (cancelled Flux)"() { - setup: - def latch = new CountDownLatch(1) - def flux = ReactorTracedMethods.traceAsyncFlux(latch) - - expect: - TEST_WRITER.size() == 0 - - when: - latch.countDown() - flux.subscribe(new ReactorTracedMethods.CancelSubscriber<>()) - - then: - assertTraces(1) { - trace(1) { - span { - resourceName "ReactorTracedMethods.traceAsyncFlux" - operationName "ReactorTracedMethods.traceAsyncFlux" - tags { - defaultTags() - "$Tags.COMPONENT" "opentelemetry" - "$Tags.SPAN_KIND" "internal" - } - } - } - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorCoreTest.groovy b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorCoreTest.groovy deleted file mode 100644 index f2654e4e1a4..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/ReactorCoreTest.groovy +++ /dev/null @@ -1,598 +0,0 @@ -import datadog.trace.agent.test.InstrumentationSpecification -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.opentelemetry.api.GlobalOpenTelemetry -import io.opentelemetry.api.trace.Span -import io.opentracing.util.GlobalTracer -import org.reactivestreams.Publisher -import org.reactivestreams.Subscriber -import org.reactivestreams.Subscription -import reactor.core.publisher.Flux -import reactor.core.publisher.Mono -import reactor.core.scheduler.Schedulers -import reactor.util.context.Context -import spock.lang.Shared - -import java.time.Duration -import java.util.concurrent.CompletableFuture - -import static datadog.trace.agent.test.utils.TraceUtils.basicSpan -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 - -class ReactorCoreTest extends InstrumentationSpecification { - - public static final String EXCEPTION_MESSAGE = "test exception" - - @Shared - def addOne = { i -> - addOneFunc(i) - } - - @Shared - def addTwo = { i -> - addTwoFunc(i) - } - - @Shared - def throwException = { - throw new RuntimeException(EXCEPTION_MESSAGE) - } - - @Override - void configurePreAgent() { - super.configurePreAgent() - injectSysConfig("dd.trace.otel.enabled", "true") - } - - def "Publisher '#name' test"() { - when: - def result = assemblePublisherUnderTrace(publisherSupplier) - - then: - result == expected - and: - assertTraces(1) { - sortSpansByStart() - trace(workSpans + 2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - - for (int i = 0; i < workSpans; i++) { - span { - resourceName "addOne" - operationName "addOne" - childOf(span(1)) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - // spotless:off - name | expected | workSpans | publisherSupplier - "basic mono" | 2 | 1 | { -> Mono.just(1).map(addOne) } - "two operations mono" | 4 | 2 | { -> Mono.just(2).map(addOne).map(addOne) } - "delayed mono" | 4 | 1 | { -> Mono.just(3).delayElement(Duration.ofMillis(100)).map(addOne) } - "delayed twice mono" | 6 | 2 | { -> - Mono.just(4).delayElement(Duration.ofMillis(100)).map(addOne).delayElement(Duration.ofMillis(100)).map(addOne) - } - "basic flux" | [6, 7] | 2 | { -> Flux.fromIterable([5, 6]).map(addOne) } - "two operations flux" | [8, 9] | 4 | { -> Flux.fromIterable([6, 7]).map(addOne).map(addOne) } - "delayed flux" | [8, 9] | 2 | { -> - Flux.fromIterable([7, 8]).delayElements(Duration.ofMillis(100)).map(addOne) - } - "delayed twice flux" | [10, 11] | 4 | { -> - Flux.fromIterable([8, 9]).delayElements(Duration.ofMillis(100)).map(addOne).delayElements(Duration.ofMillis(100)).map(addOne) - } - "mono from callable" | 12 | 2 | { -> Mono.fromCallable({ addOneFunc(10) }).map(addOne) } - // spotless:on - } - - def "Publisher error '#name' test"() { - when: - assemblePublisherUnderTrace(publisherSupplier) - - then: - def exception = thrown RuntimeException - exception.message == EXCEPTION_MESSAGE - and: - assertTraces(1) { - sortSpansByStart() - trace(2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - errored true - tags { - "$Tags.COMPONENT" "trace" - errorTags(RuntimeException, EXCEPTION_MESSAGE) - defaultTags() - } - } - - // It's important that we don't attach errors at the Reactor level so that we don't - // impact the spans on reactor integrations such as netty and lettuce, as reactor is - // more of a context propagation mechanism than something we would be tracking for - // errors this is ok. - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - } - } - - where: - name | publisherSupplier - "mono" | { -> Mono.error(new RuntimeException(EXCEPTION_MESSAGE)) } - "flux" | { -> Flux.error(new RuntimeException(EXCEPTION_MESSAGE)) } - } - - def "Publisher step '#name' test"() { - when: - assemblePublisherUnderTrace(publisherSupplier) - - then: - def exception = thrown RuntimeException - exception.message == EXCEPTION_MESSAGE - and: - assertTraces(1) { - sortSpansByStart() - trace(workSpans + 2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - errored true - tags { - "$Tags.COMPONENT" "trace" - errorTags(RuntimeException, EXCEPTION_MESSAGE) - defaultTags() - } - } - - // It's important that we don't attach errors at the Reactor level so that we don't - // impact the spans on reactor integrations such as netty and lettuce, as reactor is - // more of a context propagation mechanism than something we would be tracking for - // errors this is ok. - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - - for (int i = 0; i < workSpans; i++) { - span { - resourceName "addOne" - operationName "addOne" - childOf(span(1)) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - name | workSpans | publisherSupplier - "basic mono failure" | 1 | { -> Mono.just(1).map(addOne).map({ throwException() }) } - "basic flux failure" | 1 | { -> Flux.fromIterable([5, 6]).map(addOne).map({ throwException() }) } - } - - def "Publisher '#name' cancel"() { - when: - cancelUnderTrace(publisherSupplier) - - then: - assertTraces(1) { - trace(2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - } - } - - where: - name | publisherSupplier - "basic mono" | { -> Mono.just(1) } - "basic flux" | { -> Flux.fromIterable([5, 6]) } - } - - def "Publisher chain spans have the correct parent for '#name'"() { - when: - assemblePublisherUnderTrace(publisherSupplier) - - then: - assertTraces(1) { - trace(workSpans + 2) { - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - - for (int i = 0; i < workSpans; i++) { - span { - resourceName "addOne" - operationName "addOne" - childOf(span(1)) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - name | workSpans | publisherSupplier - "basic mono" | 3 | { -> Mono.just(1).map(addOne).map(addOne).then(Mono.just(1).map(addOne)) } - "basic flux" | 5 | { -> Flux.fromIterable([5, 6]).map(addOne).map(addOne).then(Mono.just(1).map(addOne)) } - } - - def "Publisher chain spans have the correct parents from subscription time"() { - when: - def mono = Mono.just(42) - .map(addOne) - .map(addTwo) - - runUnderTrace("trace-parent") { - mono.block() - } - - then: - assertTraces(1) { - trace(3) { - sortSpansByStart() - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - } - span { - operationName "addOne" - childOf span(0) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - span { - operationName "addTwo" - childOf span(0) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - def "Publisher chain spans have the correct parents from subscription time '#name'"() { - when: - assemblePublisherUnderTrace { - // The "add one" operations in the publisher created here should be children of the publisher-parent - Publisher publisher = publisherSupplier() - - AgentSpan intermediate = startSpan("test", "intermediate") - AgentScope scope = activateSpan(intermediate) - try { - if (publisher instanceof Mono) { - return ((Mono) publisher).map(addTwo) - } else if (publisher instanceof Flux) { - return ((Flux) publisher).map(addTwo) - } - throw new IllegalStateException("Unknown publisher type") - } finally { - intermediate.finish() - scope.close() - } - } - - then: - assertTraces(1) { - trace(3 + 2 * workItems) { - sortSpansByStart() - span { - resourceName "trace-parent" - operationName "trace-parent" - parent() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - - basicSpan(it, "publisher-parent", "publisher-parent", span(0)) - basicSpan(it, "intermediate", span(1)) - - for (int i = 0; i < 2 * workItems; i = i + 2) { - span { - operationName "addOne" - childOf span(1) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - span { - operationName "addTwo" - childOf span(1) - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - } - } - - where: - name | workItems | publisherSupplier - "basic mono" | 1 | { -> Mono.just(1).map(addOne) } - "basic flux" | 2 | { -> Flux.fromIterable([1, 2]).map(addOne) } - } - - def "Fluxes produce the right number of results on '#schedulerName' scheduler"() { - when: - List values = Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) - .parallel() - .runOn(scheduler) - .flatMap({ num -> Mono.just(num.toString() + " on " + Thread.currentThread().getName()) }) - .sequential() - .collectList() - .block() - - then: - values.size() == 4 - - where: - schedulerName | scheduler - "parallel" | Schedulers.parallel() - "elastic" | (isLatestDepTest ? Schedulers."boundedElastic"() : Schedulers."elastic"()) - "single" | Schedulers.single() - "immediate" | Schedulers.immediate() - } - - def "subscribe-time context propagates across threads with #name"() { - // Guards that the thread-confined publisher hand-off (HandoffContext) does not break cross-thread - // propagation: the @Trace "addOne" spans run in map's onNext on scheduler threads and must still - // be children of the subscribe-time parent. - when: - runUnderTrace("parent") { - pipeline.call().collectList().block() - } - - then: - assertTraces(1) { - trace(3) { - sortSpansByStart() - span { - operationName "parent" - parent() - } - span { - operationName "addOne" - childOf span(0) - } - span { - operationName "addOne" - childOf span(0) - } - } - } - - where: - name | pipeline - "publishOn" | { - Flux.just(1, 2).publishOn(Schedulers.parallel()).map(addOne) - } - "subscribeOn" | { - Flux.just(1, 2).subscribeOn(Schedulers.single()).map(addOne) - } - "subscribeOn+publishOn" | { - Flux.just(1, 2) - .subscribeOn(Schedulers.single()) - .publishOn(Schedulers.parallel()) - .map(addOne) - } - } - - def "Context propagation through reactor context with span #spanType"() { - when: - runUnderTrace("parent", { - Mono.defer { - def span = buildSpan() - Mono.just(0) - .subscriberContext(Context.of("dd.span", span)) - .doFinally { - ignored -> finishSpan(span) - } - } - .map(this::addOneFunc) - .block() - }) - then: - assertTraces(1, { - trace(3) { - basicSpan(it, "parent") - basicSpan(it, "contextual", span(0), null, ['span.kind': { - true - }]) // otel spans sets span.kind - span { - operationName "addOne" - childOfPrevious() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - }) - where: - spanType | buildSpan | finishSpan - "datadog" | { - TEST_TRACER.buildSpan("reactor-core", "contextual").start() - } | { - AgentSpan span -> span.finish() - } - "opentracing" | { - GlobalTracer.get().buildSpan("contextual").start() - } | { - io.opentracing.Span span -> span.finish() - } - "otel" | { - GlobalOpenTelemetry.get().getTracer("").spanBuilder("contextual").setAttribute("operation.name", "contextual").startSpan() - } | { - Span span -> span.end() - } - } - - def "Bad values for dd.span are tolerated"() { - when: - runUnderTrace("parent", { - Mono.just(1) - .subscriberContext(Context.of("dd.span", "Hello world")) - .map(this::addOneFunc) - .block() - }) - then: - assertTraces(1, { - trace(2) { - basicSpan(it, "parent") - span { - operationName "addOne" - childOfPrevious() - tags { - "$Tags.COMPONENT" "trace" - defaultTags() - } - } - } - }) - } - - def "test currentContext() calls on inner operator is not throwing a NPE on the advice"() { - when: - def mono = Flux.range(1, 100).windowUntil { it % 10 == 0 }.count() - then: - // we are not interested into asserting a trace structure but only that the instrumentation error count is 0 - assert mono.block() == 11 - } - - - def "span in the context has to be activated when the publisher subscribes"() { - when: - // the mono is subscribed (block) when first is active. - // However we expect that the span third will have second as parent and not first - // because we set the parent explicitly in the reactor context (dd.span key) - def result = runUnderTrace("first", { - runUnderTrace("second", { - def mono = Mono.defer { - Mono.fromCompletionStage(CompletableFuture.supplyAsync { - runUnderTrace("third", { - "hello world" - }) - }) - }.subscriberContext(Context.of("dd.span", TEST_TRACER.activeSpan())) - mono - }) - .block() - }) - then: - assert result == "hello world" - assertTraces(1, { - trace(3, true) { - basicSpan(it, "first") - basicSpan(it, "second", span(0)) - basicSpan(it, "third", span(1)) - } - }) - } - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - def assemblePublisherUnderTrace(def publisherSupplier) { - def span = startSpan("test", "publisher-parent") - // After this activation, the "add two" operations below should be children of this span - def scope = activateSpan(span) - - Publisher publisher = publisherSupplier() - try { - // Read all data from publisher - if (publisher instanceof Mono) { - return publisher.block() - } else if (publisher instanceof Flux) { - return publisher.toStream().toArray({ size -> new Integer[size] }) - } - - throw new RuntimeException("Unknown publisher: " + publisher) - } finally { - span.finish() - scope.close() - } - } - - @Trace(operationName = "trace-parent", resourceName = "trace-parent") - def cancelUnderTrace(def publisherSupplier) { - final AgentSpan span = startSpan("test", "publisher-parent") - AgentScope scope = activateSpan(span) - - def publisher = publisherSupplier() - publisher.subscribe(new Subscriber() { - void onSubscribe(Subscription subscription) { - subscription.cancel() - } - - void onNext(Integer t) { - } - - void onError(Throwable error) { - } - - void onComplete() { - } - }) - - scope.close() - span.finish() - } - - @Trace(operationName = "addOne", resourceName = "addOne") - def static addOneFunc(int i) { - return i + 1 - } - - @Trace(operationName = "addTwo", resourceName = "addTwo") - def static addTwoFunc(int i) { - return i + 2 - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/SubscriptionTest.groovy b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/SubscriptionTest.groovy deleted file mode 100644 index 41d0715e1c7..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/groovy/SubscriptionTest.groovy +++ /dev/null @@ -1,214 +0,0 @@ -import static datadog.trace.agent.test.utils.TraceUtils.basicSpan -import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace -import static datadog.trace.agent.test.utils.TraceUtils.runnableUnderTrace - -import datadog.trace.agent.test.InstrumentationSpecification -import datadog.trace.api.Trace -import datadog.trace.bootstrap.instrumentation.api.AgentTracer -import datadog.trace.bootstrap.instrumentation.api.Tags -import reactor.core.publisher.DirectProcessor -import reactor.core.publisher.EmitterProcessor -import reactor.core.publisher.Flux -import reactor.core.publisher.FluxProcessor -import reactor.core.publisher.Mono -import reactor.core.publisher.TopicProcessor -import reactor.core.publisher.WorkQueueProcessor - -import java.util.concurrent.CountDownLatch - -class SubscriptionTest extends InstrumentationSpecification { - - def "subscription test with processor #processor.class and #consumers consumers"() { - when: - def connection = (FluxProcessor) processor - CountDownLatch published = new CountDownLatch(consumers) - // need to wait for subscriber when using a non buffered processor - CountDownLatch subscribed = new CountDownLatch(consumers) - for (int i = 0; i < consumers; i++) { - def t = new Thread({ - runnableUnderTrace("parent", { - connection.subscribe { - it.query() - published.countDown() - } - subscribed.countDown() - published.await() - }) - }) - t.start() - } - subscribed.await() - connection.sink().next(new Connection()) - published.await() - then: - assertTraces(consumers) { - for (int i = 0; i < consumers; i++) { - trace(2) { - basicSpan(it, "parent") - basicSpan(it, "Connection.query", span(0)) - } - } - } - where: - processor | consumers - DirectProcessor.create() | 1 - EmitterProcessor.create() | 1 - TopicProcessor.create() | 1 - WorkQueueProcessor.create() | 1 - DirectProcessor.create() | 3 - EmitterProcessor.create() | 3 - TopicProcessor.create() | 3 - } - - def "test broadcasting flux"() { - when: - def connection = Flux. just(new Connection()).publish() - - CountDownLatch published = new CountDownLatch(1) - // need to wait for subscriber when using a non buffered processor - CountDownLatch subscribed = new CountDownLatch(1) - def t = new Thread({ - runnableUnderTrace("parent", { - connection.subscribe { - it.query() - published.countDown() - } - subscribed.countDown() - published.await() - }) - }) - t.start() - subscribed.await() - connection.connect() - published.await() - then: - assertTraces(1) { - trace(2) { - basicSpan(it, "parent") - basicSpan(it, "Connection.query", span(0)) - } - } - } - - def 'Mono then()'() { - when: - runUnderTrace("parent", { - Mono.create { - def creator = it - runUnderTrace("child1", { - creator.success() - }) - } - .then(Mono.create { - def creator = it - runUnderTrace("child2", { - creator.success() - }) - }) - .block() - }) - then: - assertTraces(1, { - trace(3) { - sortSpansByStart() - basicSpan(it, "parent") - basicSpan(it, "child1", span(0)) - basicSpan(it, "child2", span(0)) - } - }) - } - - def 'Mono with lifecycle spans and no parent'() { - when: - Mono.fromCallable { - runUnderTrace("span", { - Mono.just("Hello World") - }) - } - .doOnNext { - runUnderTrace("onNext", {}) - } - .doFinally { runUnderTrace("finally", {}) } - .doAfterTerminate { runUnderTrace("after", {}) } - .block() - then: - assertTraces(4, { - trace(1) { - basicSpan(it, "span") - } - trace(1) { - basicSpan(it, "onNext") - } - trace(1) { - basicSpan(it, "after") - } - trace(1) { - basicSpan(it, "finally") - } - }) - } - - def 'Mono with lifecycle spans and parent'() { - when: - runUnderTrace("parent", { - Mono.fromCallable { - runUnderTrace("span", { - Mono.just("Hello World") - }) - } - .doOnNext { - runUnderTrace("onNext", {}) - } - .doFinally { runUnderTrace("finally", {}) } - .doAfterTerminate { runUnderTrace("after", {}) } - .block() - }) - then: - assertTraces(1, { - trace(5) { - sortSpansByStart() - basicSpan(it, "parent") - basicSpan(it, "span", span(0)) - basicSpan(it, "onNext", span(0)) - basicSpan(it, "after", span(0)) - basicSpan(it, "finally", span(0)) - } - }) - } - - def "Optimized mono should finish attached spans"() { - when: - runUnderTrace("parent", { - optimizedTraced().block() - }) - - then: - assertTraces(1) { - trace(2) { - basicSpan(it, "parent") - span { - resourceName "SubscriptionTest.optimizedTraced" - operationName "trace.annotation" - tags { - defaultTags() - "$Tags.COMPONENT" "trace" - } - } - } - } - } - - @Trace - static Mono optimizedTraced() { - // MonoSource is optimized by not subscribing recursively - return Mono.from (Mono.just("test")) - } - - static class Connection { - static int query() { - def span = AgentTracer.startSpan("test", "Connection.query") - span.finish() - return new Random().nextInt() - } - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/annotatedsample/ReactorTracedMethods.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/annotatedsample/ReactorTracedMethods.java deleted file mode 100644 index a3450a31e20..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/annotatedsample/ReactorTracedMethods.java +++ /dev/null @@ -1,75 +0,0 @@ -package annotatedsample; - -import static java.util.concurrent.TimeUnit.SECONDS; - -import io.opentelemetry.instrumentation.annotations.WithSpan; -import java.util.concurrent.CountDownLatch; -import org.reactivestreams.Subscriber; -import org.reactivestreams.Subscription; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -public class ReactorTracedMethods { - @WithSpan - public static Mono traceAsyncMono(CountDownLatch latch) { - return Mono.fromCallable( - () -> { - await(latch); - return "hello"; - }); - } - - @WithSpan - public static Mono traceAsyncFailingMono(CountDownLatch latch, Exception exception) { - return Mono.fromCallable( - () -> { - await(latch); - throw exception; - }); - } - - @WithSpan - public static Flux traceAsyncFlux(CountDownLatch latch) { - return Flux.create( - emitter -> { - await(latch); - emitter.next("hello"); - emitter.complete(); - }); - } - - @WithSpan - public static Flux traceAsyncFailingFlux(CountDownLatch latch, Exception exception) { - return Flux.create( - emitter -> { - await(latch); - emitter.error(exception); - }); - } - - private static void await(CountDownLatch latch) { - try { - if (!latch.await(5, SECONDS)) { - throw new IllegalStateException("Latch still locked"); - } - } catch (InterruptedException e) { - throw new RuntimeException(e); - } - } - - public static class CancelSubscriber implements Subscriber { - @Override - public void onSubscribe(Subscription s) { - s.cancel(); - } - - @Override - public void onNext(T t) {} - - @Override - public void onError(Throwable t) {} - - @Override - public void onComplete() {} - } -} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java new file mode 100644 index 00000000000..17fda52e6da --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java @@ -0,0 +1,587 @@ +package testdog.trace.instrumentation.reactorcore; + +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 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 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 reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Scheduler; +import reactor.core.scheduler.Schedulers; + +class ReactorCoreTest 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 Mono) { + return ((Mono) publisher).block(); + } else if (publisher instanceof Flux) { + List list = ((Flux) publisher).collectList().block(); + 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); + + Object publisher = publisherSupplier.get(); + Flux flux; + if (publisher instanceof Mono) { + flux = ((Mono) publisher).flux(); + } else { + flux = (Flux) publisher; + } + + try { + flux.subscribe( + new org.reactivestreams.Subscriber() { + @Override + public void onSubscribe(org.reactivestreams.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 mono", + new Object[] {2}, + 1, + (Supplier) () -> Mono.just(1).map(Worker::addOne)), + Arguments.of( + "two operations mono", + new Object[] {4}, + 2, + (Supplier) () -> Mono.just(2).map(Worker::addOne).map(Worker::addOne)), + Arguments.of( + "basic flux", + new Object[] {6, 7}, + 2, + (Supplier) () -> Flux.fromIterable(Arrays.asList(5, 6)).map(Worker::addOne)), + Arguments.of( + "two operations flux", + new Object[] {8, 9}, + 4, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(6, 7)) + .map(Worker::addOne) + .map(Worker::addOne)), + Arguments.of( + "mono from callable", + new Object[] {12}, + 2, + (Supplier) + () -> Mono.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( + "mono", (Supplier) () -> Mono.error(new RuntimeException(EXCEPTION_MESSAGE))), + Arguments.of( + "flux", (Supplier) () -> Flux.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 mono failure", + 1, + (Supplier) + () -> Mono.just(1).map(Worker::addOne).map(i -> Worker.throwException())), + Arguments.of( + "basic flux failure", + 1, + (Supplier) + () -> + Flux.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 mono", (Supplier) () -> Mono.just(1)), + Arguments.of( + "basic flux", (Supplier) () -> Flux.fromIterable(Arrays.asList(5, 6)))); + } + + @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 mono", + 3, + (Supplier) + () -> + Mono.just(1) + .map(Worker::addOne) + .map(Worker::addOne) + .concatWith(Mono.just(1).map(Worker::addOne))), + Arguments.of( + "basic flux", + 5, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(5, 6)) + .map(Worker::addOne) + .map(Worker::addOne) + .concatWith(Mono.just(1).map(Worker::addOne)))); + } + + @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 (block) ---------------------- + + @Test + void correctParentsFromSubscriptionTimeBlock() { + Mono mono = Mono.just(42).map(Worker::addOne).map(Worker::addTwo); + + Worker.runUnderTraceParent( + () -> { + mono.block(); + 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 mono", 1, (Supplier) () -> Mono.just(1).map(Worker::addOne)), + Arguments.of( + "basic flux", + 2, + (Supplier) + () -> Flux.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( + () -> { + Object publisher = supplier.get(); + + AgentSpan intermediate = startSpan("test", "intermediate"); + Worker.intermediateId = intermediate.getSpanId(); + AgentScope scope = activateSpan(intermediate); + try { + if (publisher instanceof Mono) { + return ((Mono) publisher).map(Worker::addTwo); + } else if (publisher instanceof Flux) { + return ((Flux) 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("parallel", Schedulers.parallel()), + Arguments.of("elastic", Schedulers.elastic()), + Arguments.of("single", Schedulers.single())); + } + + @ParameterizedTest(name = "Flux produces the right number of results on ''{0}'' scheduler") + @MethodSource("schedulerArgs") + void schedulers(String schedulerName, Scheduler scheduler) { + List values = + Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) + .parallel() + .runOn(scheduler) + .flatMap( + num -> Mono.just(num.toString() + " on " + Thread.currentThread().getName())) + .sequential() + .collectList() + .block(); + + 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. + tracer.flush(); + assertEquals( + 0, + writer.getTraceCount(), + () -> "Unexpected traces emitted without active trace: " + writer); + } + + @ParameterizedTest(name = "Flux propagates context on ''{0}'' scheduler") + @MethodSource("schedulerArgs") + void fluxParallelContextPropagation(String schedulerName, Scheduler scheduler) { + Worker.assemblePublisherUnderTrace( + () -> + Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) + .parallel() + .runOn(scheduler) + .flatMap(num -> Mono.just(num).map(Worker::addOne)) + .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 -------------------------------- + + static List noSpuriousTracesArgs() { + return Arrays.asList( + Arguments.of( + "flux", + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) + .map(i -> i + 1) + .collectList() + .block()), + Arguments.of("mono", (Supplier) () -> Mono.just(1).map(i -> i + 1).block())); + } + + @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/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/SubscriptionTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/SubscriptionTest.java new file mode 100644 index 00000000000..b6d598d1544 --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/SubscriptionTest.java @@ -0,0 +1,130 @@ +package testdog.trace.instrumentation.reactorcore; + +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 java.util.Random; +import java.util.concurrent.CountDownLatch; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +class SubscriptionTest extends AbstractInstrumentationTest { + + @Test + void monoSubscriptionPropagatesParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Mono connection = Mono.create(sink -> sink.success(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 fluxSubscriptionPropagatesParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Flux connections = + Flux.create( + emitter -> { + emitter.next(new Connection()); + emitter.complete(); + }); + connections.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 monoJustSubscriptionPropagatesParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Mono connection = Mono.just(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 fluxFromIterableSubscriptionPropagatesParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Flux connections = + Flux.fromIterable(java.util.Arrays.asList(new Connection(), new Connection())); + connections.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"), + span().childOfIndex(0).operationName("Connection.query"))); + } + + static class Connection { + int query() { + AgentSpan span = startSpan("test", "Connection.query"); + span.finish(); + return new Random().nextInt(); + } + } +} From cf8b2872b1a5127d42396bb3be3868e36198f123 Mon Sep 17 00:00:00 2001 From: Jordan Wong Date: Tue, 14 Jul 2026 09:05:55 -0400 Subject: [PATCH 2/4] style(reactor-core-3.1): apply spotless formatting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Mechanical fix from ./gradlew :dd-java-agent:instrumentation:reactor-core-3.1:spotlessApply to unblock the spotless CI check. No functional changes. Remaining CI failures on this reference PR are preserved as intentional research signal for #11927 reviewers: - test_inst (all JVMs): Rule #2 regen-preservation violation — eval reorganized the package layout from datadog.trace.instrumentation.reactor.core (master) to datadog.trace.instrumentation.reactorcore (eval output), producing a duplicate ReactorAsyncResultExtension registration. - latestDepTest: eval's generated test references Schedulers.elastic() which was removed in Reactor 3.4+; illustrates a latestDep API-drift gap in the current skill draft. --- .../reactorcore/FluxInstrumentation.java | 3 +-- .../reactorcore/MonoInstrumentation.java | 3 +-- .../instrumentation/reactorcore/ReactorCoreModule.java | 3 +-- .../instrumentation/reactorcore/ReactorCoreTest.java | 10 +++------- 4 files changed, 6 insertions(+), 13 deletions(-) diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java index 904ddaefac8..bc6123cd74e 100644 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/FluxInstrumentation.java @@ -50,8 +50,7 @@ public static ContextScope onSubscribe( @Advice.This final Flux flux, @Advice.Argument(value = 0, readOnly = false) CoreSubscriber subscriber) { if (subscriber != null) { - Context parentContext = - InstrumentationContext.get(Flux.class, Context.class).get(flux); + Context parentContext = InstrumentationContext.get(Flux.class, Context.class).get(flux); if (parentContext != null) { subscriber = new TracingCoreSubscriber<>(subscriber, parentContext); return parentContext.attach(); diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java index 6ecc9d4cdd5..32814dea87b 100644 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/MonoInstrumentation.java @@ -50,8 +50,7 @@ public static ContextScope onSubscribe( @Advice.This final Mono mono, @Advice.Argument(value = 0, readOnly = false) CoreSubscriber subscriber) { if (subscriber != null) { - Context parentContext = - InstrumentationContext.get(Mono.class, Context.class).get(mono); + Context parentContext = InstrumentationContext.get(Mono.class, Context.class).get(mono); if (parentContext != null) { // wrap the subscriber so spans from its events treat the captured span as their parent subscriber = new TracingCoreSubscriber<>(subscriber, parentContext); diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java index 499f47b26a0..469b54ea3df 100644 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactorcore/ReactorCoreModule.java @@ -19,8 +19,7 @@ public ReactorCoreModule() { @Override public String[] helperClassNames() { return new String[] { - packageName + ".TracingCoreSubscriber", - packageName + ".ReactorAsyncResultExtension", + packageName + ".TracingCoreSubscriber", packageName + ".ReactorAsyncResultExtension", }; } diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java index 17fda52e6da..b79744111e4 100644 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java @@ -163,9 +163,7 @@ static List publisherSuccessArgs() { 4, (Supplier) () -> - Flux.fromIterable(Arrays.asList(6, 7)) - .map(Worker::addOne) - .map(Worker::addOne)), + Flux.fromIterable(Arrays.asList(6, 7)).map(Worker::addOne).map(Worker::addOne)), Arguments.of( "mono from callable", new Object[] {12}, @@ -423,8 +421,7 @@ static List subscriptionTimeIntermediateArgs() { Arguments.of( "basic flux", 2, - (Supplier) - () -> Flux.fromIterable(Arrays.asList(1, 2)).map(Worker::addOne))); + (Supplier) () -> Flux.fromIterable(Arrays.asList(1, 2)).map(Worker::addOne))); } @ParameterizedTest( @@ -505,8 +502,7 @@ void schedulers(String schedulerName, Scheduler scheduler) { Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) .parallel() .runOn(scheduler) - .flatMap( - num -> Mono.just(num.toString() + " on " + Thread.currentThread().getName())) + .flatMap(num -> Mono.just(num.toString() + " on " + Thread.currentThread().getName())) .sequential() .collectList() .block(); From cb53d9776283375485354c7986d7c06bd29cc9c7 Mon Sep 17 00:00:00 2001 From: Jordan Wong Date: Tue, 14 Jul 2026 09:09:26 -0400 Subject: [PATCH 3/4] fix(reactor-core-3.1): remove duplicate ReactorAsyncResultExtension MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The blind regen produced a new instrumentation class under 'datadog.trace.instrumentation.reactorcore.ReactorAsyncResultExtension' that duplicates the pre-existing master version at 'datadog.trace.instrumentation.reactor.core.ReactorAsyncResultExtension'. Both classes register themselves via static initializer to AsyncResultExtensions.register(), causing runtime conflict and test_inst failures across all JVM shards (17, 21, 25, tip). This commit removes the master leftover so only the eval-generated version remains. The package-rename itself (reactor.core -> reactorcore) is preserved as intentional research signal — it reflects the current PR #11927 draft skill's output and is a Rule #2 (regen-preservation) observation reviewers should see. Remaining failure preserved as expected research signal: - latestDepTest: eval uses Schedulers.elastic() which was removed in Reactor 3.4+; illustrates a latestDep API-drift gap. --- .../core/ReactorAsyncResultExtension.java | 44 ------------------- 1 file changed, 44 deletions(-) delete mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorAsyncResultExtension.java diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorAsyncResultExtension.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorAsyncResultExtension.java deleted file mode 100644 index ab16ccac7db..00000000000 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/main/java/datadog/trace/instrumentation/reactor/core/ReactorAsyncResultExtension.java +++ /dev/null @@ -1,44 +0,0 @@ -package datadog.trace.instrumentation.reactor.core; - -import datadog.trace.bootstrap.instrumentation.api.AgentSpan; -import datadog.trace.bootstrap.instrumentation.api.EagerHelper; -import datadog.trace.bootstrap.instrumentation.java.concurrent.AsyncResultExtension; -import datadog.trace.bootstrap.instrumentation.java.concurrent.AsyncResultExtensions; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; - -public class ReactorAsyncResultExtension implements AsyncResultExtension, EagerHelper { - static { - AsyncResultExtensions.register(new ReactorAsyncResultExtension()); - } - - /** - * Register the extension as an {@link AsyncResultExtension} using static class initialization. - *
- * It uses an empty static method call to ensure the class loading and the one-time-only static - * class initialization. This will ensure this extension will only be registered once under {@link - * AsyncResultExtensions}. - */ - public static void init() {} - - @Override - public boolean supports(Class result) { - return Flux.class.isAssignableFrom(result) || Mono.class.isAssignableFrom(result); - } - - @Override - public Object apply(Object result, AgentSpan span) { - if (result instanceof Flux) { - return ((Flux) result) - .doOnError(span::addThrowable) - .doOnTerminate(span::finish) - .doOnCancel(span::finish); - } else if (result instanceof Mono) { - return ((Mono) result) - .doOnError(span::addThrowable) - .doOnTerminate(span::finish) - .doOnCancel(span::finish); - } - return null; - } -} From 3f485e671ef7776cee4326af37e08454dfa174fa Mon Sep 17 00:00:00 2001 From: Jordan Wong Date: Tue, 14 Jul 2026 09:33:27 -0400 Subject: [PATCH 4/4] fix(reactor-core-3.1): drop Schedulers.elastic() from scheduler test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The generated ReactorCoreTest.schedulerArgs() included a case using Schedulers.elastic(), which was deprecated in Reactor 3.4 and removed in 3.5+. This blocked both :check (via aggregate test compile) and :latestDepTest (via direct API drift) across all JVM CI shards. Dropped the "elastic" row from the parameterized args, leaving "parallel" and "single" (both present in Reactor 3.1+ and current latestDep). Preserves test coverage of scheduler-hop trace propagation without depending on a removed API. Research observation preserved in the PR body: this failure mode is a latestDep API-drift gap in the current skill draft — master handles version-sensitive tests by splitting them into a separate `latestDepTest` source set (see master's src/latestDepTest/groovy/ReactorCoreTest.groovy). --- .../trace/instrumentation/reactorcore/ReactorCoreTest.java | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java index b79744111e4..8dad9536b0f 100644 --- a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactorcore/ReactorCoreTest.java @@ -489,9 +489,14 @@ void correctParentsFromSubscriptionTime(String name, int workItems, Supplier schedulerArgs() { + // Note: Schedulers.elastic() was deprecated in Reactor 3.4 and removed in 3.5+, so + // it was omitted here to keep the base test set compilable against both the module's + // declared min (3.1) and the latestDep version. See the PR body for the research + // observation about how the toolkit could handle latestDep API-drift better — + // master splits Reactor version-sensitive tests into a separate `latestDepTest` + // source set. return Arrays.asList( Arguments.of("parallel", Schedulers.parallel()), - Arguments.of("elastic", Schedulers.elastic()), Arguments.of("single", Schedulers.single())); }