RxJava concurrency test
MODIFY
instrumentation/rxjava/rxjava-2.0/testing/src/main/groovy/io/opentelemetry/instrumentation/rxjava2/AbstractRxJava2Test.groovy
instrumentation/rxjava/rxjava-2.0/testing/src/main/groovy/io/opentelemetry/instrumentation/rxjava2/AbstractRxJava2Test.groovy
package io.opentelemetry.instrumentation.rxjava2
+import io.opentelemetry.api.common.AttributeKey
+
import static io.opentelemetry.instrumentation.test.utils.TraceUtils.basicSpan
import static io.opentelemetry.instrumentation.test.utils.TraceUtils.runUnderTrace
import static io.opentelemetry.instrumentation.test.utils.TraceUtils.runUnderTraceWithoutExceptionCatch
@@ -317,6 +319,44 @@ abstract class AbstractRxJava2Test extends InstrumentationSpecification {
scheduler << [Schedulers.newThread(), Schedulers.computation(), Schedulers.single(), Schedulers.trampoline()]
}
+ def "test many ongoing trace chains on '#scheduler'"() {
+ setup:
+ int iterations = 100
+ Set<Long> remainingIterations = new HashSet<>((0L..(iterations - 1)).toList())
+
+ when:
+ RxJava2ConcurrencyTestHelper.launchAndWait(scheduler, iterations, 60000)
+
+ then:
+ assertTraces(iterations) {
+ for (int i = 0; i < iterations; i++) {
+ trace(i, 3) {
+ long iteration = -1
+ span(0) {
+ name("outer")
+ iteration = span.getAttributes().get(AttributeKey.longKey("iteration")).toLong()
+ assert remainingIterations.remove(iteration)
+ }
+ span(1) {
+ name("middle")
+ childOf(span(0))
+ assert span.getAttributes().get(AttributeKey.longKey("iteration")) == iteration
+ }
+ span(2) {
+ name("inner")
+ childOf(span(1))
+ assert span.getAttributes().get(AttributeKey.longKey("iteration")) == iteration
+ }
+ }
+ }
+ }
+
+ assert remainingIterations.isEmpty()
+
+ where:
+ scheduler << [Schedulers.newThread(), Schedulers.computation(), Schedulers.single(), Schedulers.trampoline()]
+ }
+
def cancelUnderTrace(def publisherSupplier) {
runUnderTraceWithoutExceptionCatch("publisher-parent") {
def publisher = publisherSupplier()

ADD
instrumentation/rxjava/rxjava-2.0/testing/src/main/groovy/io/opentelemetry/instrumentation/rxjava2/RxJava2ConcurrencyTestHelper.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.rxjava2;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.instrumentation.test.utils.TraceUtils;
+import io.reactivex.Scheduler;
+import io.reactivex.Single;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This test creates the specified number of traces with three spans: 1) Outer (root) span 2) Middle
+ * span, child of outer, created in success handler of the chain subscribed to in the context of the
+ * outer span (with some delay and map thrown in for good measure) 3) Inner span, child of middle,
+ * created in the success handler of a new chain started and subscribed to in the the middle span
+ *
+ * <p>The varying delays between the stages where each span is created should guarantee that
+ * scheduler threads handling various stages of the chain will have to alternate between contexts
+ * from different traces.
+ */
+public class RxJava2ConcurrencyTestHelper {
+ public static void launchAndWait(Scheduler scheduler, int iterations, long timeoutMillis) {
+ CountDownLatch latch = new CountDownLatch(iterations);
+
+ for (int i = 0; i < iterations; i++) {
+ launchOuter(new Iteration(scheduler, latch, i));
+ }
+
+ try {
+ // Continue even on timeout so the test assertions can show what is missing
+ //noinspection ResultOfMethodCallIgnored
+ latch.await(timeoutMillis, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void launchOuter(Iteration iteration) {
+ TraceUtils.runUnderTrace(
+ "outer",
+ () -> {
+ Span.current().setAttribute("iteration", iteration.index);
+
+ Single.fromCallable(() -> iteration)
+ .subscribeOn(iteration.scheduler)
+ .observeOn(iteration.scheduler)
+ // Use varying delay so that different stages of the chain would alternate.
+ .delay(iteration.index % 10, TimeUnit.MILLISECONDS, iteration.scheduler)
+ .map((it) -> it)
+ .delay(iteration.index % 10, TimeUnit.MILLISECONDS, iteration.scheduler)
+ .doOnSuccess(RxJava2ConcurrencyTestHelper::launchInner)
+ .subscribe();
+
+ return null;
+ });
+ }
+
+ private static void launchInner(Iteration iteration) {
+ TraceUtils.runUnderTrace(
+ "middle",
+ () -> {
+ Span.current().setAttribute("iteration", iteration.index);
+
+ Single.fromCallable(() -> iteration)
+ .subscribeOn(iteration.scheduler)
+ .observeOn(iteration.scheduler)
+ .delay(iteration.index % 10, TimeUnit.MILLISECONDS, iteration.scheduler)
+ .doOnSuccess(
+ (it) -> {
+ TraceUtils.runUnderTrace(
+ "inner",
+ () -> {
+ Span.current().setAttribute("iteration", it.index);
+ return null;
+ });
+ it.countDown.countDown();
+ })
+ .subscribe();
+
+ return null;
+ });
+ }
+
+ private static class Iteration {
+ public final Scheduler scheduler;
+ public final CountDownLatch countDown;
+ public final int index;
+
+ private Iteration(Scheduler scheduler, CountDownLatch countDown, int index) {
+ this.scheduler = scheduler;
+ this.countDown = countDown;
+ this.index = index;
+ }
+ }
+}

MODIFY
instrumentation/rxjava/rxjava-3.0/testing/src/main/groovy/io/opentelemetry/instrumentation/rxjava3/AbstractRxJava3Test.groovy
instrumentation/rxjava/rxjava-3.0/testing/src/main/groovy/io/opentelemetry/instrumentation/rxjava3/AbstractRxJava3Test.groovy
package io.opentelemetry.instrumentation.rxjava3
+import io.opentelemetry.api.common.AttributeKey
import io.reactivex.rxjava3.core.BackpressureStrategy
import io.reactivex.rxjava3.core.Completable
import io.reactivex.rxjava3.core.Flowable
@@ -318,6 +319,44 @@ abstract class AbstractRxJava3Test extends InstrumentationSpecification {
scheduler << [Schedulers.newThread(), Schedulers.computation(), Schedulers.single(), Schedulers.trampoline()]
}
+ def "test many ongoing trace chains on '#scheduler'"() {
+ setup:
+ int iterations = 100
+ Set<Long> remainingIterations = new HashSet<>((0L..(iterations - 1)).toList())
+
+ when:
+ RxJava3ConcurrencyTestHelper.launchAndWait(scheduler, iterations, 60000)
+
+ then:
+ assertTraces(iterations) {
+ for (int i = 0; i < iterations; i++) {
+ trace(i, 3) {
+ long iteration = -1
+ span(0) {
+ name("outer")
+ iteration = span.getAttributes().get(AttributeKey.longKey("iteration")).toLong()
+ assert remainingIterations.remove(iteration)
+ }
+ span(1) {
+ name("middle")
+ childOf(span(0))
+ assert span.getAttributes().get(AttributeKey.longKey("iteration")) == iteration
+ }
+ span(2) {
+ name("inner")
+ childOf(span(1))
+ assert span.getAttributes().get(AttributeKey.longKey("iteration")) == iteration
+ }
+ }
+ }
+ }
+
+ assert remainingIterations.isEmpty()
+
+ where:
+ scheduler << [Schedulers.newThread(), Schedulers.computation(), Schedulers.single(), Schedulers.trampoline()]
+ }
+
def cancelUnderTrace(def publisherSupplier) {
runUnderTraceWithoutExceptionCatch("publisher-parent") {
def publisher = publisherSupplier()

ADD
instrumentation/rxjava/rxjava-3.0/testing/src/main/groovy/io/opentelemetry/instrumentation/rxjava3/RxJava3ConcurrencyTestHelper.java
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.instrumentation.rxjava3;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.instrumentation.test.utils.TraceUtils;
+import io.reactivex.rxjava3.core.Scheduler;
+import io.reactivex.rxjava3.core.Single;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * This test creates the specified number of traces with three spans: 1) Outer (root) span 2) Middle
+ * span, child of outer, created in success handler of the chain subscribed to in the context of the
+ * outer span (with some delay and map thrown in for good measure) 3) Inner span, child of middle,
+ * created in the success handler of a new chain started and subscribed to in the the middle span
+ *
+ * <p>The varying delays between the stages where each span is created should guarantee that
+ * scheduler threads handling various stages of the chain will have to alternate between contexts
+ * from different traces.
+ */
+public class RxJava3ConcurrencyTestHelper {
+ public static void launchAndWait(Scheduler scheduler, int iterations, long timeoutMillis) {
+ CountDownLatch latch = new CountDownLatch(iterations);
+
+ for (int i = 0; i < iterations; i++) {
+ launchOuter(new Iteration(scheduler, latch, i));
+ }
+
+ try {
+ // Continue even on timeout so the test assertions can show what is missing
+ //noinspection ResultOfMethodCallIgnored
+ latch.await(timeoutMillis, TimeUnit.MILLISECONDS);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ private static void launchOuter(Iteration iteration) {
+ TraceUtils.runUnderTrace(
+ "outer",
+ () -> {
+ Span.current().setAttribute("iteration", iteration.index);
+
+ Single.fromCallable(() -> iteration)
+ .subscribeOn(iteration.scheduler)
+ .observeOn(iteration.scheduler)
+ // Use varying delay so that different stages of the chain would alternate.
+ .delay(iteration.index % 10, TimeUnit.MILLISECONDS, iteration.scheduler)
+ .map((it) -> it)
+ .delay(iteration.index % 10, TimeUnit.MILLISECONDS, iteration.scheduler)
+ .doOnSuccess(RxJava3ConcurrencyTestHelper::launchInner)
+ .subscribe();
+
+ return null;
+ });
+ }
+
+ private static void launchInner(Iteration iteration) {
+ TraceUtils.runUnderTrace(
+ "middle",
+ () -> {
+ Span.current().setAttribute("iteration", iteration.index);
+
+ Single.fromCallable(() -> iteration)
+ .subscribeOn(iteration.scheduler)
+ .observeOn(iteration.scheduler)
+ .delay(iteration.index % 10, TimeUnit.MILLISECONDS, iteration.scheduler)
+ .doOnSuccess(
+ (it) -> {
+ TraceUtils.runUnderTrace(
+ "inner",
+ () -> {
+ Span.current().setAttribute("iteration", it.index);
+ return null;
+ });
+ it.countDown.countDown();
+ })
+ .subscribe();
+
+ return null;
+ });
+ }
+
+ private static class Iteration {
+ public final Scheduler scheduler;
+ public final CountDownLatch countDown;
+ public final int index;
+
+ private Iteration(Scheduler scheduler, CountDownLatch countDown, int index) {
+ this.scheduler = scheduler;
+ this.countDown = countDown;
+ this.index = index;
+ }
+ }
+}

RxJava concurrency test (#3036)