Skip to content

Commit 35096e2

Browse files
amarzialiValentinZakharovdevflow.devflow-routing-intake
authored
Add RxJava 3 instrumentation (#11506)
Scaffold rxjava-3.0 instrumentation module Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Add rxjava-3.0 Tracing observer/subscriber wrappers Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Add rxjava-3.0 type instrumentations, module and async result extension Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Register rxjava3 async result extension for Graal native-image Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Add rxjava-3.0 SubscriptionTest Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Add rxjava-3.0 core context propagation test Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Fix pending-trace leak from RxJava3 AbstractDirectTask static initializer RxJava 3 introduced io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask, whose static initializer constructs FINISHED/DISPOSED sentinel FutureTask instances. When that initializer first runs under an active trace (e.g. the first delay/timeout scheduler hop inside a span), the executor instrumentation captures a ScopeContinuation on those static singletons that is never cancelled, leaking the pending trace so it is never reported. Disable async propagation while AbstractDirectTask's type initializer runs, mirroring the existing reactor.core.scheduler.SchedulerTask/WorkerTask handling. The matcher is inert unless RxJava 3 is on the classpath. RxJava 2 has no equivalent class. Restores the delayed-Maybe coverage in RxJava3Test, which fails without this fix and passes with it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Add rxjava-3.0 @WithSpan async result extension test Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Add rxjava-3.0 Java 8 interop context-propagation investigation test Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Unify rxjava-3.0 test package under testdog.trace.instrumentation.rxjava3 Two tests were under datadog.trace.instrumentation.rxjava3 while the @Trace-using tests must live under testdog.* (the agent ignores datadog.* for instrumentation). Move all four tests to testdog.* for consistency. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> review Add config json entry Skip root context capture in Observable and Single CaptureParentSpanAdvice Add scheduler context propagation and no-spurious-traces tests for Observable and Single Fix misleading comment in correctParentsFromSubscriptionTime test Extend cancel test to cover all five reactive types Ensure publisher-parent span is always finished in cancelUnderTrace Normalize PropagateParentSpanAdvice guard and add @nonnull to TracingObserver spotless spotless Extend SubscriptionTest to cover all five RxJava 3 reactive types Add cancelledNever test to cover dispose-path span lifecycle in RxJava3ResultExtensionTest Co-authored-by: ValentinZakharov <valentin.zakharov@datadoghq.com> Co-authored-by: devflow.devflow-routing-intake <devflow.devflow-routing-intake@kubernetes.us1.ddbuild.io>
1 parent 265cec1 commit 35096e2

22 files changed

Lines changed: 2125 additions & 0 deletions

File tree

‎dd-java-agent/instrumentation/graal/graal-native-image-20.0/src/main/java/datadog/trace/instrumentation/graal/nativeimage/NativeImageGeneratorRunnerInstrumentation.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,7 @@ public static void onEnter(@Advice.Argument(value = 0, readOnly = false) String[
157157
+ "datadog.trace.instrumentation.reactivestreams.ReactiveStreamsAsyncResultExtension:build_time,"
158158
+ "datadog.trace.instrumentation.reactor.core.ReactorAsyncResultExtension:build_time,"
159159
+ "datadog.trace.instrumentation.rxjava2.RxJavaAsyncResultExtension:build_time,"
160+
+ "datadog.trace.instrumentation.rxjava3.RxJavaAsyncResultExtension:build_time,"
160161
+ "datadog.trace.logging.ddlogger.DDLogger:build_time,"
161162
+ "datadog.trace.logging.ddlogger.DDLoggerFactory:build_time,"
162163
+ "datadog.trace.logging.ddlogger.DDLoggerFactory$HelperWrapper:build_time,"

‎dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-1.8/src/main/java/datadog/trace/instrumentation/java/concurrent/AsyncPropagatingDisableInstrumentation.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,17 @@ public AsyncPropagatingDisableInstrumentation() {
4747
namedOneOf("reactor.core.scheduler.SchedulerTask", "reactor.core.scheduler.WorkerTask");
4848
private static final ElementMatcher<TypeDescription> RXJAVA2_DISABLED_TYPE_INITIALIZERS =
4949
named("io.reactivex.internal.schedulers.AbstractDirectTask");
50+
51+
/**
52+
* RxJava 3's AbstractDirectTask creates FINISHED/DISPOSED sentinel FutureTask instances in its
53+
* static initializer. If that initializer runs while a trace is active (e.g. the first scheduled
54+
* delay/timeout under a span), the executor instrumentation captures a continuation on those
55+
* static singletons that is never cancelled, leaking the pending trace. Disable async propagation
56+
* while the type initializer runs.
57+
*/
58+
private static final ElementMatcher<TypeDescription> RXJAVA3_DISABLED_TYPE_INITIALIZERS =
59+
named("io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask");
60+
5061
private static final ElementMatcher<TypeDescription> NETTY_GLOBAL_EVENT_EXECUTOR =
5162
namedOneOf(
5263
"io.netty.util.concurrent.GlobalEventExecutor",
@@ -90,6 +101,7 @@ public String[] knownMatchingTypes() {
90101
"org.apache.activemq.broker.TransactionBroker",
91102
"com.mongodb.internal.connection.DefaultConnectionPool$AsyncWorkManager",
92103
"io.reactivex.internal.schedulers.AbstractDirectTask",
104+
"io.reactivex.rxjava3.internal.schedulers.AbstractDirectTask",
93105
"jdk.internal.net.http.HttpClientImpl",
94106
LETTUCE_HANDSHAKE_HANDLER,
95107
"io.netty.util.concurrent.GlobalEventExecutor",
@@ -110,6 +122,7 @@ public ElementMatcher<TypeDescription> hierarchyMatcher() {
110122
.or(GRPC_MANAGED_CHANNEL)
111123
.or(REACTOR_DISABLED_TYPE_INITIALIZERS)
112124
.or(RXJAVA2_DISABLED_TYPE_INITIALIZERS)
125+
.or(RXJAVA3_DISABLED_TYPE_INITIALIZERS)
113126
.or(JAVA_HTTP_CLIENT);
114127
}
115128

@@ -196,6 +209,8 @@ public void methodAdvice(MethodTransformer transformer) {
196209
isTypeInitializer().and(isDeclaredBy(REACTOR_DISABLED_TYPE_INITIALIZERS)), advice);
197210
transformer.applyAdvice(
198211
isTypeInitializer().and(isDeclaredBy(RXJAVA2_DISABLED_TYPE_INITIALIZERS)), advice);
212+
transformer.applyAdvice(
213+
isTypeInitializer().and(isDeclaredBy(RXJAVA3_DISABLED_TYPE_INITIALIZERS)), advice);
199214
transformer.applyAdvice(
200215
isTypeInitializer().and(isDeclaredBy(NETTY_GLOBAL_EVENT_EXECUTOR)), advice);
201216
transformer.applyAdvice(namedOneOf("sendAsync").and(isDeclaredBy(JAVA_HTTP_CLIENT)), advice);

‎dd-java-agent/instrumentation/rxjava/rxjava-3.0/build.gradle‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
muzzle {
2+
pass {
3+
group = "io.reactivex.rxjava3"
4+
module = "rxjava"
5+
versions = "[3.0.0,)"
6+
}
7+
// Assert the rxjava3 advice never resolves against rxjava2 — the two namespaces
8+
// must not overlap. rxjava3 references io.reactivex.rxjava3.core.*, absent from the
9+
// rxjava2 artifact, so muzzle must fail to match it.
10+
fail {
11+
name = "rxjava2-must-not-match"
12+
group = "io.reactivex.rxjava2"
13+
module = "rxjava"
14+
versions = "[2.0.0,)"
15+
}
16+
}
17+
18+
apply from: "$rootDir/gradle/java.gradle"
19+
20+
addTestSuiteForDir('latestDepTest', 'test')
21+
22+
dependencies {
23+
compileOnly group: 'org.reactivestreams', name: 'reactive-streams', version: '1.0.3'
24+
compileOnly group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0'
25+
26+
testImplementation project(':dd-java-agent:instrumentation:datadog:tracing:trace-annotation')
27+
testImplementation project(':dd-java-agent:instrumentation:opentelemetry:opentelemetry-annotations-1.20')
28+
testImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '3.0.0'
29+
testImplementation group: 'io.opentelemetry.instrumentation', name: 'opentelemetry-instrumentation-annotations', version: '1.28.0'
30+
31+
// Load the rxjava2 instrumenter at test runtime to prove the two versions coexist on
32+
// the agent without interference (it stays dormant with only rxjava3 on the classpath).
33+
testRuntimeOnly project(':dd-java-agent:instrumentation:rxjava:rxjava-2.0')
34+
35+
latestDepTestImplementation group: 'io.reactivex.rxjava3', name: 'rxjava', version: '+'
36+
}

‎dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/CompletableInstrumentation.java‎

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
package datadog.trace.instrumentation.rxjava3;
2+
3+
import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
4+
import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
5+
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
6+
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
7+
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
8+
9+
import datadog.context.Context;
10+
import datadog.context.ContextScope;
11+
import datadog.trace.agent.tooling.Instrumenter;
12+
import datadog.trace.bootstrap.InstrumentationContext;
13+
import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
14+
import io.reactivex.rxjava3.core.Completable;
15+
import io.reactivex.rxjava3.core.CompletableObserver;
16+
import net.bytebuddy.asm.Advice;
17+
18+
public final class CompletableInstrumentation
19+
implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
20+
21+
@Override
22+
public String instrumentedType() {
23+
return "io.reactivex.rxjava3.core.Completable";
24+
}
25+
26+
@Override
27+
public void methodAdvice(MethodTransformer transformer) {
28+
transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
29+
transformer.applyAdvice(
30+
isMethod()
31+
.and(named("subscribe"))
32+
.and(takesArguments(1))
33+
.and(takesArgument(0, named("io.reactivex.rxjava3.core.CompletableObserver"))),
34+
getClass().getName() + "$PropagateParentSpanAdvice");
35+
}
36+
37+
public static class CaptureParentSpanAdvice {
38+
@Advice.OnMethodExit(suppress = Throwable.class)
39+
public static void onConstruct(@Advice.This final Completable completable) {
40+
Context parentContext = Java8BytecodeBridge.getCurrentContext();
41+
if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
42+
InstrumentationContext.get(Completable.class, Context.class)
43+
.put(completable, parentContext);
44+
}
45+
}
46+
}
47+
48+
public static class PropagateParentSpanAdvice {
49+
@Advice.OnMethodEnter(suppress = Throwable.class)
50+
public static ContextScope onSubscribe(
51+
@Advice.This final Completable completable,
52+
@Advice.Argument(value = 0, readOnly = false) CompletableObserver observer) {
53+
if (observer != null) {
54+
Context parentContext =
55+
InstrumentationContext.get(Completable.class, Context.class).get(completable);
56+
if (parentContext != null) {
57+
// wrap the observer so spans from its events treat the captured span as their parent
58+
observer = new TracingCompletableObserver(observer, parentContext);
59+
// attach the context here in case additional observers are created during subscribe
60+
return parentContext.attach();
61+
}
62+
}
63+
return null;
64+
}
65+
66+
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
67+
public static void closeScope(@Advice.Enter final ContextScope scope) {
68+
if (scope != null) {
69+
scope.close();
70+
}
71+
}
72+
}
73+
}

‎dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/FlowableInstrumentation.java‎

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
package datadog.trace.instrumentation.rxjava3;
2+
3+
import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
4+
import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
5+
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
6+
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
7+
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
8+
9+
import datadog.context.Context;
10+
import datadog.context.ContextScope;
11+
import datadog.trace.agent.tooling.Instrumenter;
12+
import datadog.trace.bootstrap.InstrumentationContext;
13+
import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
14+
import io.reactivex.rxjava3.core.Flowable;
15+
import io.reactivex.rxjava3.core.FlowableSubscriber;
16+
import net.bytebuddy.asm.Advice;
17+
18+
public final class FlowableInstrumentation
19+
implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
20+
21+
@Override
22+
public String instrumentedType() {
23+
return "io.reactivex.rxjava3.core.Flowable";
24+
}
25+
26+
@Override
27+
public void methodAdvice(MethodTransformer transformer) {
28+
transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
29+
transformer.applyAdvice(
30+
isMethod()
31+
.and(named("subscribe"))
32+
.and(takesArguments(1))
33+
.and(takesArgument(0, named("io.reactivex.rxjava3.core.FlowableSubscriber"))),
34+
getClass().getName() + "$PropagateParentSpanAdvice");
35+
}
36+
37+
public static class CaptureParentSpanAdvice {
38+
@Advice.OnMethodExit(suppress = Throwable.class)
39+
public static void onConstruct(@Advice.This final Flowable<?> flowable) {
40+
Context parentContext = Java8BytecodeBridge.getCurrentContext();
41+
if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
42+
InstrumentationContext.get(Flowable.class, Context.class).put(flowable, parentContext);
43+
}
44+
}
45+
}
46+
47+
public static class PropagateParentSpanAdvice {
48+
@Advice.OnMethodEnter(suppress = Throwable.class)
49+
public static ContextScope onSubscribe(
50+
@Advice.This final Flowable<?> flowable,
51+
@Advice.Argument(value = 0, readOnly = false) FlowableSubscriber<?> subscriber) {
52+
if (subscriber != null) {
53+
Context parentContext =
54+
InstrumentationContext.get(Flowable.class, Context.class).get(flowable);
55+
if (parentContext != null) {
56+
// wrap the subscriber so spans from its events treat the captured span as their parent
57+
subscriber = new TracingSubscriber<>(subscriber, parentContext);
58+
// attach the context here in case additional subscribers are created during subscribe
59+
return parentContext.attach();
60+
}
61+
}
62+
return null;
63+
}
64+
65+
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
66+
public static void closeScope(@Advice.Enter final ContextScope scope) {
67+
if (scope != null) {
68+
scope.close();
69+
}
70+
}
71+
}
72+
}

‎dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/MaybeInstrumentation.java‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
package datadog.trace.instrumentation.rxjava3;
2+
3+
import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
4+
import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
5+
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
6+
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
7+
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
8+
9+
import datadog.context.Context;
10+
import datadog.context.ContextScope;
11+
import datadog.trace.agent.tooling.Instrumenter;
12+
import datadog.trace.bootstrap.InstrumentationContext;
13+
import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
14+
import io.reactivex.rxjava3.core.Maybe;
15+
import io.reactivex.rxjava3.core.MaybeObserver;
16+
import net.bytebuddy.asm.Advice;
17+
18+
public final class MaybeInstrumentation
19+
implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
20+
@Override
21+
public String instrumentedType() {
22+
return "io.reactivex.rxjava3.core.Maybe";
23+
}
24+
25+
@Override
26+
public void methodAdvice(MethodTransformer transformer) {
27+
transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
28+
transformer.applyAdvice(
29+
isMethod()
30+
.and(named("subscribe"))
31+
.and(takesArguments(1))
32+
.and(takesArgument(0, named("io.reactivex.rxjava3.core.MaybeObserver"))),
33+
getClass().getName() + "$PropagateParentSpanAdvice");
34+
}
35+
36+
public static class CaptureParentSpanAdvice {
37+
@Advice.OnMethodExit(suppress = Throwable.class)
38+
public static void onConstruct(@Advice.This final Maybe<?> maybe) {
39+
Context parentContext = Java8BytecodeBridge.getCurrentContext();
40+
if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
41+
InstrumentationContext.get(Maybe.class, Context.class).put(maybe, parentContext);
42+
}
43+
}
44+
}
45+
46+
public static class PropagateParentSpanAdvice {
47+
@Advice.OnMethodEnter(suppress = Throwable.class)
48+
public static ContextScope onSubscribe(
49+
@Advice.This final Maybe<?> maybe,
50+
@Advice.Argument(value = 0, readOnly = false) MaybeObserver<?> observer) {
51+
if (observer != null) {
52+
Context parentContext = InstrumentationContext.get(Maybe.class, Context.class).get(maybe);
53+
if (parentContext != null) {
54+
// wrap the observer so spans from its events treat the captured span as their parent
55+
observer = new TracingMaybeObserver<>(observer, parentContext);
56+
// attach the context here in case additional observers are created during subscribe
57+
return parentContext.attach();
58+
}
59+
}
60+
return null;
61+
}
62+
63+
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
64+
public static void closeScope(@Advice.Enter final ContextScope scope) {
65+
if (scope != null) {
66+
scope.close();
67+
}
68+
}
69+
}
70+
}

‎dd-java-agent/instrumentation/rxjava/rxjava-3.0/src/main/java/datadog/trace/instrumentation/rxjava3/ObservableInstrumentation.java‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
package datadog.trace.instrumentation.rxjava3;
2+
3+
import static datadog.trace.agent.tooling.bytebuddy.matcher.NameMatchers.named;
4+
import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
5+
import static net.bytebuddy.matcher.ElementMatchers.isMethod;
6+
import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
7+
import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
8+
9+
import datadog.context.Context;
10+
import datadog.context.ContextScope;
11+
import datadog.trace.agent.tooling.Instrumenter;
12+
import datadog.trace.bootstrap.InstrumentationContext;
13+
import datadog.trace.bootstrap.instrumentation.api.Java8BytecodeBridge;
14+
import io.reactivex.rxjava3.core.Observable;
15+
import io.reactivex.rxjava3.core.Observer;
16+
import net.bytebuddy.asm.Advice;
17+
18+
public final class ObservableInstrumentation
19+
implements Instrumenter.ForSingleType, Instrumenter.HasMethodAdvice {
20+
@Override
21+
public String instrumentedType() {
22+
return "io.reactivex.rxjava3.core.Observable";
23+
}
24+
25+
@Override
26+
public void methodAdvice(MethodTransformer transformer) {
27+
transformer.applyAdvice(isConstructor(), getClass().getName() + "$CaptureParentSpanAdvice");
28+
transformer.applyAdvice(
29+
isMethod()
30+
.and(named("subscribe"))
31+
.and(takesArguments(1))
32+
.and(takesArgument(0, named("io.reactivex.rxjava3.core.Observer"))),
33+
getClass().getName() + "$PropagateParentSpanAdvice");
34+
}
35+
36+
public static class CaptureParentSpanAdvice {
37+
@Advice.OnMethodExit(suppress = Throwable.class)
38+
public static void onConstruct(@Advice.This final Observable<?> observable) {
39+
Context parentContext = Java8BytecodeBridge.getCurrentContext();
40+
if (parentContext != null && parentContext != Java8BytecodeBridge.getRootContext()) {
41+
InstrumentationContext.get(Observable.class, Context.class).put(observable, parentContext);
42+
}
43+
}
44+
}
45+
46+
public static class PropagateParentSpanAdvice {
47+
@Advice.OnMethodEnter(suppress = Throwable.class)
48+
public static ContextScope onSubscribe(
49+
@Advice.This final Observable<?> observable,
50+
@Advice.Argument(value = 0, readOnly = false) Observer<?> observer) {
51+
if (observer != null) {
52+
Context parentContext =
53+
InstrumentationContext.get(Observable.class, Context.class).get(observable);
54+
if (parentContext != null) {
55+
// wrap the observer so spans from its events treat the captured span as their parent
56+
observer = new TracingObserver<>(observer, parentContext);
57+
// attach the context here in case additional observers are created during subscribe
58+
return parentContext.attach();
59+
}
60+
}
61+
return null;
62+
}
63+
64+
@Advice.OnMethodExit(onThrowable = Throwable.class, suppress = Throwable.class)
65+
public static void closeScope(@Advice.Enter final ContextScope scope) {
66+
if (scope != null) {
67+
scope.close();
68+
}
69+
}
70+
}
71+
}

0 commit comments

Comments
 (0)