-
Notifications
You must be signed in to change notification settings - Fork 342
Add RxJava 3 instrumentation #11506
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Merged
gh-worker-dd-mergequeue-cf854d
merged 22 commits into
master
from
andrea.marziali/rxjava3
Jun 26, 2026
Merged
Add RxJava 3 instrumentation #11506
Changes from all commits
Commits
Show all changes
22 commits
Select commit
Hold shift + click to select a range
98d7069
Scaffold rxjava-3.0 instrumentation module
amarziali b6ab351
Add rxjava-3.0 Tracing observer/subscriber wrappers
amarziali 598d379
Add rxjava-3.0 type instrumentations, module and async result extension
amarziali 70b71d4
Register rxjava3 async result extension for Graal native-image
amarziali f53cd41
Add rxjava-3.0 SubscriptionTest
amarziali c918b1b
Add rxjava-3.0 core context propagation test
amarziali ecb8e0c
Fix pending-trace leak from RxJava3 AbstractDirectTask static initial…
amarziali 80fc14c
Add rxjava-3.0 @WithSpan async result extension test
amarziali fd3ed43
Add rxjava-3.0 Java 8 interop context-propagation investigation test
amarziali f819795
Unify rxjava-3.0 test package under testdog.trace.instrumentation.rxj…
amarziali 1cb0608
review
amarziali 9d8e4fc
Add config json entry
amarziali b400ab3
Skip root context capture in Observable and Single CaptureParentSpanA…
ValentinZakharov eb21c8d
Add scheduler context propagation and no-spurious-traces tests for Ob…
ValentinZakharov 5356f12
Fix misleading comment in correctParentsFromSubscriptionTime test
ValentinZakharov 998eee9
Extend cancel test to cover all five reactive types
ValentinZakharov 33a9de4
Ensure publisher-parent span is always finished in cancelUnderTrace
ValentinZakharov 97dc359
Normalize PropagateParentSpanAdvice guard and add @Nonnull to Tracing…
ValentinZakharov 5d68e49
spotless
ValentinZakharov af33b38
spotless
ValentinZakharov 3ebbb91
Extend SubscriptionTest to cover all five RxJava 3 reactive types
ValentinZakharov a1843ed
Add cancelledNever test to cover dispose-path span lifecycle in RxJav…
ValentinZakharov File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
36 changes: 36 additions & 0 deletions
36
dd-java-agent/instrumentation/rxjava/rxjava-3.0/build.gradle
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| muzzle { | ||
| pass { | ||
| group = "io.reactivex.rxjava3" | ||
| module = "rxjava" | ||
| versions = "[3.0.0,)" | ||
| } | ||
| // Assert the rxjava3 advice never resolves against rxjava2 — the two namespaces | ||
| // must not overlap. rxjava3 references io.reactivex.rxjava3.core.*, absent from the | ||
| // rxjava2 artifact, so muzzle must fail to match it. | ||
| fail { | ||
| name = "rxjava2-must-not-match" | ||
| group = "io.reactivex.rxjava2" | ||
| module = "rxjava" | ||
| versions = "[2.0.0,)" | ||
| } | ||
| } | ||
|
|
||
| apply from: "$rootDir/gradle/java.gradle" | ||
|
|
||
| addTestSuiteForDir('latestDepTest', 'test') | ||
|
|
||
| dependencies { | ||
| compileOnly group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.3' | ||
| compileOnly group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0' | ||
|
|
||
| testImplementation project(':dd-java-agent:instrumentation:datadog:tracing:trace-annotation') | ||
| testImplementation project(':dd-java-agent:instrumentation:opentelemetry:opentelemetry-annotations-1.20') | ||
| testImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0' | ||
| testImplementation group: 'io.opentelemetry.instrumentation', name: 'opentelemetry-instrumentation-annotations', version: '1.28.0' | ||
|
|
||
| // Load the rxjava2 instrumenter at test runtime to prove the two versions coexist on | ||
| // the agent without interference (it stays dormant with only rxjava3 on the classpath). | ||
| testRuntimeOnly project(':dd-java-agent:instrumentation:rxjava:rxjava-2.0') | ||
|
|
||
| latestDepTestImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '+' | ||
| } |
73 changes: 73 additions & 0 deletions
73
...a-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/CompletableInstrumentation.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,73 @@ | ||
| package datadog.trace.instrumentation.rxjava3; | ||
|
|
||
| import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; | ||
| 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 datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge; | ||
| import io.reactivex.rxjava3.core.Completable; | ||
| import io.reactivex.rxjava3.core.CompletableObserver; | ||
| import net.bytebuddy.asm.Advice; | ||
|
|
||
| public final class CompletableInstrumentation | ||
| implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { | ||
|
|
||
| @Override | ||
| public String instrumentedType() { | ||
| return "io.reactivex.rxjava3.core.Completable"; | ||
| } | ||
|
|
||
| @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("io.reactivex.rxjava3.core.CompletableObserver"))), | ||
| getClass().getName() + "$PropagateParentSpanAdvice"); | ||
| } | ||
|
|
||
| public static class CaptureParentSpanAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static void onConstruct(@Advice.This final Completable completable) { | ||
| Context parentContext = Java8BytecodeBridge.getCurrentContext(); | ||
| if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) { | ||
| InstrumentationContext.get(Completable.class, Context.class) | ||
| .put(completable, parentContext); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public static class PropagateParentSpanAdvice { | ||
| @Advice.OnMethodEnter(suppress = Throwable.class) | ||
| public static ContextScope onSubscribe( | ||
| @Advice.This final Completable completable, | ||
| @Advice.Argument(value = 0, readOnly = false) CompletableObserver observer) { | ||
| if (observer != null) { | ||
| Context parentContext = | ||
| InstrumentationContext.get(Completable.class, Context.class).get(completable); | ||
| if (parentContext != null) { | ||
| // wrap the observer so spans from its events treat the captured span as their parent | ||
| observer = new TracingCompletableObserver(observer, parentContext); | ||
| // attach the context here in case additional observers are created during subscribe | ||
| return parentContext.attach(); | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) | ||
| public static void closeScope(@Advice.Enter final ContextScope scope) { | ||
| if (scope != null) { | ||
| scope.close(); | ||
| } | ||
| } | ||
| } | ||
| } |
72 changes: 72 additions & 0 deletions
72
...java-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/FlowableInstrumentation.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| package datadog.trace.instrumentation.rxjava3; | ||
|
|
||
| import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; | ||
| 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 datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge; | ||
| import io.reactivex.rxjava3.core.Flowable; | ||
| import io.reactivex.rxjava3.core.FlowableSubscriber; | ||
| import net.bytebuddy.asm.Advice; | ||
|
|
||
| public final class FlowableInstrumentation | ||
| implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { | ||
|
|
||
| @Override | ||
| public String instrumentedType() { | ||
| return "io.reactivex.rxjava3.core.Flowable"; | ||
| } | ||
|
|
||
| @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("io.reactivex.rxjava3.core.FlowableSubscriber"))), | ||
| getClass().getName() + "$PropagateParentSpanAdvice"); | ||
| } | ||
|
|
||
| public static class CaptureParentSpanAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static void onConstruct(@Advice.This final Flowable<?> flowable) { | ||
| Context parentContext = Java8BytecodeBridge.getCurrentContext(); | ||
| if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) { | ||
| InstrumentationContext.get(Flowable.class, Context.class).put(flowable, parentContext); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public static class PropagateParentSpanAdvice { | ||
| @Advice.OnMethodEnter(suppress = Throwable.class) | ||
| public static ContextScope onSubscribe( | ||
| @Advice.This final Flowable<?> flowable, | ||
| @Advice.Argument(value = 0, readOnly = false) FlowableSubscriber<?> subscriber) { | ||
| if (subscriber != null) { | ||
| Context parentContext = | ||
| InstrumentationContext.get(Flowable.class, Context.class).get(flowable); | ||
| if (parentContext != null) { | ||
| // wrap the subscriber so spans from its events treat the captured span as their parent | ||
| subscriber = new TracingSubscriber<>(subscriber, parentContext); | ||
| // attach the context here in case additional subscribers are created during subscribe | ||
| return parentContext.attach(); | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) | ||
| public static void closeScope(@Advice.Enter final ContextScope scope) { | ||
| if (scope != null) { | ||
| scope.close(); | ||
| } | ||
| } | ||
| } | ||
| } | ||
70 changes: 70 additions & 0 deletions
70
.../rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/MaybeInstrumentation.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package datadog.trace.instrumentation.rxjava3; | ||
|
|
||
| import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; | ||
| 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 datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge; | ||
| import io.reactivex.rxjava3.core.Maybe; | ||
| import io.reactivex.rxjava3.core.MaybeObserver; | ||
| import net.bytebuddy.asm.Advice; | ||
|
|
||
| public final class MaybeInstrumentation | ||
| implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { | ||
| @Override | ||
| public String instrumentedType() { | ||
| return "io.reactivex.rxjava3.core.Maybe"; | ||
| } | ||
|
|
||
| @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("io.reactivex.rxjava3.core.MaybeObserver"))), | ||
| getClass().getName() + "$PropagateParentSpanAdvice"); | ||
| } | ||
|
|
||
| public static class CaptureParentSpanAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static void onConstruct(@Advice.This final Maybe<?> maybe) { | ||
| Context parentContext = Java8BytecodeBridge.getCurrentContext(); | ||
| if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) { | ||
| InstrumentationContext.get(Maybe.class, Context.class).put(maybe, parentContext); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public static class PropagateParentSpanAdvice { | ||
| @Advice.OnMethodEnter(suppress = Throwable.class) | ||
| public static ContextScope onSubscribe( | ||
| @Advice.This final Maybe<?> maybe, | ||
| @Advice.Argument(value = 0, readOnly = false) MaybeObserver<?> observer) { | ||
| if (observer != null) { | ||
| Context parentContext = InstrumentationContext.get(Maybe.class, Context.class).get(maybe); | ||
| if (parentContext != null) { | ||
| // wrap the observer so spans from its events treat the captured span as their parent | ||
| observer = new TracingMaybeObserver<>(observer, parentContext); | ||
| // attach the context here in case additional observers are created during subscribe | ||
| return parentContext.attach(); | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) | ||
| public static void closeScope(@Advice.Enter final ContextScope scope) { | ||
| if (scope != null) { | ||
| scope.close(); | ||
| } | ||
| } | ||
| } | ||
| } |
71 changes: 71 additions & 0 deletions
71
...va-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/ObservableInstrumentation.java
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| package datadog.trace.instrumentation.rxjava3; | ||
|
|
||
| import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named; | ||
| 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 datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge; | ||
| import io.reactivex.rxjava3.core.Observable; | ||
| import io.reactivex.rxjava3.core.Observer; | ||
| import net.bytebuddy.asm.Advice; | ||
|
|
||
| public final class ObservableInstrumentation | ||
| implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice { | ||
| @Override | ||
| public String instrumentedType() { | ||
| return "io.reactivex.rxjava3.core.Observable"; | ||
| } | ||
|
|
||
| @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("io.reactivex.rxjava3.core.Observer"))), | ||
| getClass().getName() + "$PropagateParentSpanAdvice"); | ||
| } | ||
|
|
||
| public static class CaptureParentSpanAdvice { | ||
| @Advice.OnMethodExit(suppress = Throwable.class) | ||
| public static void onConstruct(@Advice.This final Observable<?> observable) { | ||
| Context parentContext = Java8BytecodeBridge.getCurrentContext(); | ||
| if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) { | ||
| InstrumentationContext.get(Observable.class, Context.class).put(observable, parentContext); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| public static class PropagateParentSpanAdvice { | ||
| @Advice.OnMethodEnter(suppress = Throwable.class) | ||
| public static ContextScope onSubscribe( | ||
| @Advice.This final Observable<?> observable, | ||
| @Advice.Argument(value = 0, readOnly = false) Observer<?> observer) { | ||
| if (observer != null) { | ||
| Context parentContext = | ||
| InstrumentationContext.get(Observable.class, Context.class).get(observable); | ||
| if (parentContext != null) { | ||
| // wrap the observer so spans from its events treat the captured span as their parent | ||
| observer = new TracingObserver<>(observer, parentContext); | ||
| // attach the context here in case additional observers are created during subscribe | ||
| return parentContext.attach(); | ||
| } | ||
| } | ||
| return null; | ||
| } | ||
|
|
||
| @Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class) | ||
| public static void closeScope(@Advice.Enter final ContextScope scope) { | ||
| if (scope != null) { | ||
| scope.close(); | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When a
Flowableis consumed through the Reactive Streams API or with a subscriber that only implementsorg.reactivestreams.Subscriber, RxJava calls the separatesubscribe(Subscriber)overload and applies its own compliance wrapper for non-FlowableSubscriberinstances. This advice only matches theFlowableSubscriberoverload, soTracingSubscriberis never inserted for those subscribers and async emissions such asdelay/observeOnlose the captured parent context. This also falls short of the RxJava 2 instrumentation, which matchesorg.reactivestreams.Subscriber.Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This concern applies to RxJava 2, but not RxJava 3. In RxJava 3, subscribe(Subscriber) always delegates to subscribe(FlowableSubscriber) - either directly or by wrapping in StrictSubscriber, which implements FlowableSubscriber. Both paths converge at the same hook point. Verified against Flowable 3.1.12 bytecode