Skip to content
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.

Sign up for GitHub

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

Allow scanning instrumented reactor publishers and only allow registe… #5755

Merged
merged 5 commits into from
Apr 7, 2022
Merged

Allow scanning instrumented reactor publishers and only allow registe… #5755

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6c8d719
Allow scanning instrumented reactor publishers and only allow registe…
anuraaga Apr 6, 2022
af7c270
Avoid atomicboolean
anuraaga Apr 6, 2022
1d58ad2
Clean
anuraaga Apr 6, 2022
a42b025
Merge branch 'main' of github.com:open-telemetry/opentelemetry-java-i…
anuraaga Apr 7, 2022
add3ce7
Bug
anuraaga Apr 7, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 55 additions & 17 deletions ...ry/src/main/java/io/opentelemetry/instrumentation/reactor/ContextPropagationOperator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@
import io.opentelemetry.instrumentation.api.annotation.support.async.AsyncOperationEndStrategies;
import java.util.function.BiFunction;
import java.util.function.Function;
import javax.annotation.Nullable;
import org.reactivestreams.Publisher;
import reactor.core.CoreSubscriber;
import reactor.core.Fuseable;
Expand All @@ -39,6 +40,8 @@
/** Based on Spring Sleuth's Reactor instrumentation. */
public final class ContextPropagationOperator {

private static final Object VALUE = new Object();

public static ContextPropagationOperator create() {
return builder().build();
}
Expand Down Expand Up @@ -98,17 +101,22 @@ public static Context getOpenTelemetryContext(
* reactive stream. This should generally be called in a static initializer block in your
* application.
*/
public void registerOnEachOperator() {
public synchronized void registerOnEachOperator() {
Copy link
Contributor Author

@anuraaga anuraaga Apr 6, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When trying to use AtomicBoolean I got this bizzare exception in KotlinCoroutinesInstrumentationTest. Synchronizing for these should be fine anyways, but is it an agent bug?

[otel.javaagent 2022-04-06 07:40:51:716 +0000] [otel-javaagent-transform-safe-logger] DEBUG io.opentelemetry.javaagent.tooling.AgentInstaller$TransformLoggingListener - Failed to handle io.opentelemetry.instrumentation.reactor.ContextPropagationOperator for transformation on classloader jdk.internal.loader.ClassLoaders$AppClassLoader@7ad041f3
java.lang.IllegalStateException: Cannot assign private static final java.util.concurrent.atomic.AtomicBoolean io.opentelemetry.instrumentation.reactor.ContextPropagationOperator.enabled to boolean
	at net.bytebuddy.asm.Advice$OffsetMapping$ForField.resolve(Advice.java:2317)
	at net.bytebuddy.asm.Advice$Dispatcher$Inlining$Resolved$ForMethodEnter.doApply(Advice.java:8684)
	at net.bytebuddy.asm.Advice$Dispatcher$Inlining$Resolved$ForMethodEnter$WithDiscardedEnterType.doApply(Advice.java:8786)
	at net.bytebuddy.asm.Advice$Dispatcher$Inlining$Resolved$ForMethodEnter.apply(Advice.java:8642)
	at net.bytebuddy.asm.Advice$Dispatcher$Inlining$Resolved$AdviceMethodInliner.visitMethod(Advice.java:8332)
	at org.objectweb.asm.ClassReader.readMethod(ClassReader.java:1353)
	at org.objectweb.asm.ClassReader.accept(ClassReader.java:744)
	at org.objectweb.asm.ClassReader.accept(ClassReader.java:424)
	at net.bytebuddy.asm.Advice$Dispatcher$Inlining$Resolved$AdviceMethodInliner.apply(Advice.java:8325)
	at net.bytebuddy.asm.Advice$AdviceVisitor.onAfterExceptionTable(Advice.java:10579)
	at net.bytebuddy.utility.visitor.ExceptionTableSensitiveMethodVisitor.considerEndOfExceptionTable(ExceptionTableSensitiveMethodVisitor.java:49)
	at net.bytebuddy.utility.visitor.ExceptionTableSensitiveMethodVisitor.visitLabel(ExceptionTableSensitiveMethodVisitor.java:81)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah, probably coming from

opentelemetry-java-instrumentation/instrumentation/reactor/reactor-3.1/javaagent/src/main/java/io/opentelemetry/javaagent/instrumentation/reactor/operator/ContextPropagationOperatorInstrumentation.java

Line 109 in cd725ec

@Advice.FieldValue(value = "enabled", readOnly = false) boolean enabled) {

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

since it's a public class, better to avoid synchronizing on this?

if (enabled) {
return;
}
Hooks.onEachOperator(TracingSubscriber.class.getName(), tracingLift(asyncOperationEndStrategy));
AsyncOperationEndStrategies.instance().registerStrategy(asyncOperationEndStrategy);
enabled = true;
}

/** Unregisters the hook registered by {@link #registerOnEachOperator()}. */
public void resetOnEachOperator() {
public synchronized void resetOnEachOperator() {
if (!enabled) {
return;
}
Hooks.resetOnEachOperator(TracingSubscriber.class.getName());
AsyncOperationEndStrategies.instance().unregisterStrategy(asyncOperationEndStrategy);
enabled = false;
}

private static <T> Function<? super Publisher<T>, ? extends Publisher<T>> tracingLift(
Expand All @@ -125,8 +133,7 @@ public static <T> Mono<T> runWithContext(Mono<T> publisher, Context tracingConte
// this hack forces 'publisher' to run in the onNext callback of `TracingSubscriber`
// (created for this publisher) and with current() span that refers to span created here
// without the hack, publisher runs in the onAssembly stage, before traceContext is made current
return ScalarPropagatingMono.INSTANCE
.flatMap(i -> publisher)
return ScalarPropagatingMono.create(publisher)
.subscriberContext(ctx -> storeOpenTelemetryContext(ctx, tracingContext));
}

Expand All @@ -139,8 +146,7 @@ public static <T> Flux<T> runWithContext(Flux<T> publisher, Context tracingConte
// this hack forces 'publisher' to run in the onNext callback of `TracingSubscriber`
// (created for this publisher) and with current() span that refers to span created here
// without the hack, publisher runs in the onAssembly stage, before traceContext is made current
return ScalarPropagatingFlux.INSTANCE
.flatMap(i -> publisher)
return ScalarPropagatingFlux.create(publisher)
.subscriberContext(ctx -> storeOpenTelemetryContext(ctx, tracingContext));
}

Expand Down Expand Up @@ -177,29 +183,61 @@ static void subscribeInActiveSpan(CoreSubscriber<? super Object> actual, Object
}
}

static class ScalarPropagatingMono extends Mono<Object> {
public static final Mono<Object> INSTANCE = new ScalarPropagatingMono();
static class ScalarPropagatingMono extends Mono<Object> implements Scannable {

private final Object value = new Object();
static <T> Mono<T> create(Mono<T> source) {
return new ScalarPropagatingMono(source).flatMap(unused -> source);
}

private ScalarPropagatingMono() {}
private final Mono<?> source;

private ScalarPropagatingMono(Mono<?> source) {
this.source = source;
}

@Override
public void subscribe(CoreSubscriber<? super Object> actual) {
subscribeInActiveSpan(actual, value);
subscribeInActiveSpan(actual, VALUE);
}

@Override
@Nullable
// Interface method doesn't have type parameter so we can't add it either.
@SuppressWarnings("rawtypes")
public Object scanUnsafe(Attr attr) {
if (attr == Attr.PARENT) {
return source;
}
return null;
}
}

static class ScalarPropagatingFlux extends Flux<Object> {
public static final Flux<Object> INSTANCE = new ScalarPropagatingFlux();
static class ScalarPropagatingFlux extends Flux<Object> implements Scannable {

private final Object value = new Object();
static <T> Flux<T> create(Flux<T> source) {
return new ScalarPropagatingFlux(source).flatMap(unused -> source);
}

private ScalarPropagatingFlux() {}
private final Flux<?> source;

private ScalarPropagatingFlux(Flux<?> source) {
this.source = source;
}

@Override
public void subscribe(CoreSubscriber<? super Object> actual) {
subscribeInActiveSpan(actual, value);
subscribeInActiveSpan(actual, VALUE);
}

@Override
@Nullable
// Interface method doesn't have type parameter so we can't add it either.
@SuppressWarnings("rawtypes")
public Object scanUnsafe(Scannable.Attr attr) {
if (attr == Scannable.Attr.PARENT) {
return source;
}
return null;
}
}
}
49 changes: 39 additions & 10 deletions ...r-3.1/library/src/test/java/io/opentelemetry/instrumentation/reactor/ReactorCoreTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,10 @@
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import reactor.core.Scannable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.UnicastProcessor;

class ReactorCoreTest extends AbstractReactorCoreTest {

Expand Down Expand Up @@ -69,17 +71,16 @@ void monoInNonBlockingPublisherAssembly() {

@Test
void fluxInNonBlockingPublisherAssembly() {
Flux<Integer> source =
Flux.defer(
() -> {
Span.current().setAttribute("inner", "foo");
return Flux.just(5, 6);
});
testing.runWithSpan(
"parent",
() ->
ContextPropagationOperator.ScalarPropagatingFlux.INSTANCE
.flatMap(
unused ->
Flux.defer(
() -> {
Span.current().setAttribute("inner", "foo");
return Flux.just(5, 6);
}))
ContextPropagationOperator.ScalarPropagatingFlux.create(source)
.doOnEach(
signal -> {
if (signal.isOnError()) {
Expand Down Expand Up @@ -199,9 +200,37 @@ void noTracingBeforeRegistration() {
trace -> trace.hasSpansSatisfyingExactly(span -> span.hasName("after").hasNoParent()));
}

@Test
void monoParentsAccessible() {
UnicastProcessor<String> source = UnicastProcessor.create();
Mono<String> mono =
ContextPropagationOperator.runWithContext(source.singleOrEmpty(), Context.root());

source.onNext("foo");
source.onComplete();

assertThat(mono.block()).isEqualTo("foo");

assertThat(((Scannable) mono).parents().filter(UnicastProcessor.class::isInstance).findFirst())
.isPresent();
}

@Test
void fluxParentsAccessible() {
UnicastProcessor<String> source = UnicastProcessor.create();
Flux<String> mono = ContextPropagationOperator.runWithContext(source, Context.root());
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nit

Suggested change
Flux<String> mono = ContextPropagationOperator.runWithContext(source, Context.root());
Flux<String> flux = ContextPropagationOperator.runWithContext(source, Context.root());


source.onNext("foo");
source.onComplete();

assertThat(mono.collectList().block()).containsExactly("foo");

assertThat(((Scannable) mono).parents().filter(UnicastProcessor.class::isInstance).findFirst())
.isPresent();
}

private <T> Mono<T> monoSpan(Mono<T> mono, String spanName) {
return ContextPropagationOperator.ScalarPropagatingMono.INSTANCE
.flatMap(unused -> mono)
return ContextPropagationOperator.ScalarPropagatingMono.create(mono)
.doOnEach(
signal -> {
if (signal.isOnError()) {
Expand Down