-
Notifications
You must be signed in to change notification settings - Fork 342
Add RxJava 3 instrumentation #11849
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
Merged
Add RxJava 3 instrumentation #11849
Changes from all commits
Commits
Show all changes
29 commits
Select commit
Hold shift + click to select a range
dcdd946
Scaffold rxjava-3.0 instrumentation module
amarziali 49cf92d
Add rxjava-3.0 Tracing observer/subscriber wrappers
amarziali 55ad198
Add rxjava-3.0 type instrumentations, module and async result extension
amarziali 1f65614
Register rxjava3 async result extension for Graal native-image
amarziali 49c7174
Add rxjava-3.0 SubscriptionTest
amarziali c21e594
Add rxjava-3.0 core context propagation test
amarziali c6750de
Fix pending-trace leak from RxJava3 AbstractDirectTask static initial…
amarziali 52d4976
Add rxjava-3.0 @WithSpan async result extension test
amarziali 3724055
Add rxjava-3.0 Java 8 interop context-propagation investigation test
amarziali 5d93104
Unify rxjava-3.0 test package under testdog.trace.instrumentation.rxj…
amarziali 93a4fb2
review
amarziali e290336
Add config json entry
amarziali b4cf310
Skip root context capture in Observable and Single CaptureParentSpanA…
ValentinZakharov 23e7443
Add scheduler context propagation and no-spurious-traces tests for Ob…
ValentinZakharov fcd0e5b
Fix misleading comment in correctParentsFromSubscriptionTime test
ValentinZakharov 9734820
Extend cancel test to cover all five reactive types
ValentinZakharov af25c55
Ensure publisher-parent span is always finished in cancelUnderTrace
ValentinZakharov e54fba3
Normalize PropagateParentSpanAdvice guard and add @Nonnull to Tracing…
ValentinZakharov 647452a
spotless
ValentinZakharov 1978e70
spotless
ValentinZakharov 5ad8d82
Extend SubscriptionTest to cover all five RxJava 3 reactive types
ValentinZakharov 7677d50
Add cancelledNever test to cover dispose-path span lifecycle in RxJav…
ValentinZakharov 1ddb984
Remove strictTraceWrites(false) overrides — all tests pass under stri…
ValentinZakharov 0edde12
Remove redundant package comment — testdog convention is implicit acr…
ValentinZakharov d7f2a8c
Remove redundant comments that describe what the code does rather tha…
ValentinZakharov 14f51b0
Use childOfIndex(0) instead of childOf(parentId) in RxJava3InteropTest
ValentinZakharov 3ea2d88
Remove redundant @WithConfig for opentelemetry-annotations-1.20 integ…
ValentinZakharov 90abb3f
Remove redundant null check on parentContext in CaptureParentSpanAdvice
ValentinZakharov a6cb915
Extract local variable for repeated Context class name in contextStore()
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: '+' | ||
| } |
72 changes: 72 additions & 0 deletions
72
...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,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.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 != 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) { | ||
| 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(); | ||
| } | ||
| } | ||
| } | ||
| } |
71 changes: 71 additions & 0 deletions
71
...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,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.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 != 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) { | ||
| 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(); | ||
| } | ||
| } | ||
| } | ||
| } |
69 changes: 69 additions & 0 deletions
69
.../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,69 @@ | ||
| 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 != 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) { | ||
| 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(); | ||
| } | ||
| } | ||
| } | ||
| } |
70 changes: 70 additions & 0 deletions
70
...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,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.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 != 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) { | ||
| 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.
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.
🧹 chore: We don't want LLM reasoning here, here is a trimmed version to keep relevant info only:
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.
Java docs was cleared and simplified