From b4283decbc12415c1e1f75921801eccdddaa78bd Mon Sep 17 00:00:00 2001 From: Jordan Wong Date: Thu, 16 Jul 2026 12:08:57 -0400 Subject: [PATCH] eval: reactor-core-3.1 iteration-2 regen against skill-async-iteration-2 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reference output from the toolkit's automated instrumentation-authoring workflow, run against the iteration-2 skill draft (feat/skill-async-iteration-2, 5 skill commits on top of master). Iteration-2 was scoped to test whether the skill's new library-native context-map guidance closes the primary iteration-1 gap: dropped ReactorContextBridge + 4 supporting classes that broke Spring WebFlux + Spring Kafka reactive. Not intended to be merged as-is. Base: master @ 05671ce3b0 (iteration-2 skill under test: de407b52c2) Cost: \$7.67, wall time ~1h 6m, reviewer approved with 0 todos remaining. Diff scope: 3 net-new Java test files under src/test/java/testdog/trace/instrumentation/reactor/core/. All other generated content (production classes, build.gradle, latestDepTest source set) matches master byte-identically — the skill's Rule #2 preservation strengthenings and library-native context-map prescriptions produced a faithful regeneration of the full 7-class master surface, including ReactorContextBridge and the 4 supporting operator instrumentations that iteration-1 dropped. Score against iteration-1 gaps: all 6 CLOSED (RI-4 package layout, RI-5 enumerate master classes, RI-6 library-native context-map pattern, RI-7 latestDepTest source set, NR-5 test-scope build.gradle deps, NR-6 wrap placement). See docs/eval-research/runs/reactor-core/attempt2/outcome.md in apm-instrumentation-toolkit for the full score table. --- .../core/ReactorAsyncResultExtensionTest.java | 167 +++++ .../reactor/core/ReactorCoreTest.java | 671 ++++++++++++++++++ .../reactor/core/SubscriptionTest.java | 324 +++++++++ 3 files changed, 1162 insertions(+) create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorAsyncResultExtensionTest.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorCoreTest.java create mode 100644 dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/SubscriptionTest.java diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorAsyncResultExtensionTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorAsyncResultExtensionTest.java new file mode 100644 index 00000000000..f20581ffe8f --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorAsyncResultExtensionTest.java @@ -0,0 +1,167 @@ +package testdog.trace.instrumentation.reactor.core; + +import static datadog.trace.agent.test.assertions.Matchers.validates; +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; +import static datadog.trace.agent.test.assertions.TagsMatcher.error; +import static datadog.trace.agent.test.assertions.TagsMatcher.tag; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import annotatedsample.ReactorTracedMethods; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.assertions.SpanMatcher; +import datadog.trace.agent.test.assertions.TagsMatcher; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import datadog.trace.junit.utils.config.WithConfig; +import java.util.concurrent.CountDownLatch; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; + +@WithConfig(key = "trace.otel.enabled", value = "true") +class ReactorAsyncResultExtensionTest extends AbstractInstrumentationTest { + + static final String EXCEPTION_MESSAGE = "Test exception"; + + // The COMPONENT and SPAN_KIND tags are stored as UTF8BytesString, so we compare by string content + // rather than using is("...") which would fail the asymmetric String#equals(UTF8BytesString) + // check. + static TagsMatcher otelComponent() { + return tag(Tags.COMPONENT, validates(o -> "opentelemetry".equals(String.valueOf(o)))); + } + + static TagsMatcher internalSpanKind() { + return tag(Tags.SPAN_KIND, validates(o -> Tags.SPAN_KIND_INTERNAL.equals(String.valueOf(o)))); + } + + // The operation and resource names are stored as UTF8BytesString, so we compare by string content + // (CharSequence equality is asymmetric: String#equals(UTF8BytesString) is false). + static SpanMatcher otelSpan(String name) { + return span() + .operationName(java.util.regex.Pattern.compile(java.util.regex.Pattern.quote(name))) + .resourceName((CharSequence cs) -> name.contentEquals(cs)); + } + + // --- Mono success -------------------------------------------------------- + + @Test + void withSpanAnnotatedAsyncMono() { + CountDownLatch latch = new CountDownLatch(1); + Mono mono = ReactorTracedMethods.traceAsyncMono(latch); + + // The span must not be finished before the async result completes. + assertEquals(0, writer.size()); + + latch.countDown(); + mono.block(); + + assertTraces( + trace( + otelSpan("ReactorTracedMethods.traceAsyncMono") + .tags(defaultTags(), otelComponent(), internalSpanKind()))); + } + + // --- Mono failure -------------------------------------------------------- + + @Test + void withSpanAnnotatedAsyncFailingMono() { + CountDownLatch latch = new CountDownLatch(1); + IllegalStateException expectedException = new IllegalStateException(EXCEPTION_MESSAGE); + Mono mono = ReactorTracedMethods.traceAsyncFailingMono(latch, expectedException); + + assertEquals(0, writer.size()); + + latch.countDown(); + assertThrows(IllegalStateException.class, mono::block); + + assertTraces( + trace( + otelSpan("ReactorTracedMethods.traceAsyncFailingMono") + .error() + .tags( + defaultTags(), + otelComponent(), + internalSpanKind(), + error(IllegalStateException.class, EXCEPTION_MESSAGE)))); + } + + // --- Mono cancellation --------------------------------------------------- + + @Test + void withSpanAnnotatedAsyncCancelledMono() { + CountDownLatch latch = new CountDownLatch(1); + Mono mono = ReactorTracedMethods.traceAsyncMono(latch); + + assertEquals(0, writer.size()); + + latch.countDown(); + mono.subscribe(new ReactorTracedMethods.CancelSubscriber()); + + assertTraces( + trace( + otelSpan("ReactorTracedMethods.traceAsyncMono") + .tags(defaultTags(), otelComponent(), internalSpanKind()))); + } + + // --- Flux success -------------------------------------------------------- + + @Test + void withSpanAnnotatedAsyncFlux() { + CountDownLatch latch = new CountDownLatch(1); + Flux flux = ReactorTracedMethods.traceAsyncFlux(latch); + + assertEquals(0, writer.size()); + + latch.countDown(); + flux.blockLast(); + + assertTraces( + trace( + otelSpan("ReactorTracedMethods.traceAsyncFlux") + .tags(defaultTags(), otelComponent(), internalSpanKind()))); + } + + // --- Flux failure -------------------------------------------------------- + + @Test + void withSpanAnnotatedAsyncFailingFlux() { + CountDownLatch latch = new CountDownLatch(1); + IllegalStateException expectedException = new IllegalStateException(EXCEPTION_MESSAGE); + Flux flux = ReactorTracedMethods.traceAsyncFailingFlux(latch, expectedException); + + assertEquals(0, writer.size()); + + latch.countDown(); + assertThrows(IllegalStateException.class, flux::blockLast); + + assertTraces( + trace( + otelSpan("ReactorTracedMethods.traceAsyncFailingFlux") + .error() + .tags( + defaultTags(), + otelComponent(), + internalSpanKind(), + error(IllegalStateException.class, EXCEPTION_MESSAGE)))); + } + + // --- Flux cancellation --------------------------------------------------- + + @Test + void withSpanAnnotatedAsyncCancelledFlux() { + CountDownLatch latch = new CountDownLatch(1); + Flux flux = ReactorTracedMethods.traceAsyncFlux(latch); + + assertEquals(0, writer.size()); + + latch.countDown(); + flux.subscribe(new ReactorTracedMethods.CancelSubscriber()); + + assertTraces( + trace( + otelSpan("ReactorTracedMethods.traceAsyncFlux") + .tags(defaultTags(), otelComponent(), internalSpanKind()))); + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorCoreTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorCoreTest.java new file mode 100644 index 00000000000..76697e866ff --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/ReactorCoreTest.java @@ -0,0 +1,671 @@ +package testdog.trace.instrumentation.reactor.core; + +import static datadog.trace.agent.test.assertions.Matchers.validates; +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TagsMatcher.defaultTags; +import static datadog.trace.agent.test.assertions.TagsMatcher.error; +import static datadog.trace.agent.test.assertions.TagsMatcher.tag; +import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activeSpan; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.assertions.SpanMatcher; +import datadog.trace.agent.test.assertions.TagsMatcher; +import datadog.trace.api.Trace; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.Tags; +import java.time.Duration; +import java.util.Arrays; +import java.util.List; +import java.util.function.Supplier; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.reactivestreams.Publisher; +import org.reactivestreams.Subscriber; +import org.reactivestreams.Subscription; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.core.scheduler.Schedulers; + +class ReactorCoreTest extends AbstractInstrumentationTest { + + static final String EXCEPTION_MESSAGE = "test exception"; + + // The component tag is stored as a UTF8BytesString, so we compare by string content rather than + // using is("trace") which would fail the asymmetric String#equals(UTF8BytesString) check. + static TagsMatcher componentTrace() { + return tag(Tags.COMPONENT, validates(o -> "trace".equals(String.valueOf(o)))); + } + + static class Worker { + static long traceParentId; + static long publisherParentId; + static long intermediateId; + + static int addOne(int i) { + return addOneTraced(i); + } + + @Trace(operationName = "addOne", resourceName = "addOne") + static int addOneTraced(int i) { + return i + 1; + } + + static int addTwo(int i) { + return addTwoTraced(i); + } + + @Trace(operationName = "addTwo", resourceName = "addTwo") + static int addTwoTraced(int i) { + return i + 2; + } + + static Object throwException() { + throw new RuntimeException(EXCEPTION_MESSAGE); + } + + @Trace(operationName = "trace-parent", resourceName = "trace-parent") + @SuppressWarnings("unchecked") + static Object assemblePublisherUnderTrace(Supplier publisherSupplier) { + traceParentId = activeSpan().getSpanId(); + AgentSpan span = startSpan("test", "publisher-parent"); + publisherParentId = span.getSpanId(); + AgentScope scope = activateSpan(span); + + Publisher publisher = (Publisher) publisherSupplier.get(); + try { + if (publisher instanceof Mono) { + return ((Mono) publisher).block(); + } else if (publisher instanceof Flux) { + return ((Flux) publisher).collectList().block().toArray(new Integer[0]); + } + throw new RuntimeException("Unknown publisher: " + publisher); + } finally { + span.finish(); + scope.close(); + } + } + + @Trace(operationName = "trace-parent", resourceName = "trace-parent") + static void cancelUnderTrace(Supplier publisherSupplier) { + traceParentId = activeSpan().getSpanId(); + AgentSpan span = startSpan("test", "publisher-parent"); + publisherParentId = span.getSpanId(); + AgentScope scope = activateSpan(span); + + Publisher publisher = (Publisher) publisherSupplier.get(); + try { + publisher.subscribe( + new Subscriber() { + @Override + public void onSubscribe(Subscription subscription) { + subscription.cancel(); + } + + @Override + public void onNext(Object t) {} + + @Override + public void onError(Throwable error) {} + + @Override + public void onComplete() {} + }); + } finally { + scope.close(); + span.finish(); + } + } + + @Trace(operationName = "trace-parent", resourceName = "trace-parent") + static Object runUnderTraceParent(Supplier work) { + traceParentId = activeSpan().getSpanId(); + return work.get(); + } + } + + // --- Publisher success --------------------------------------------------- + + static List publisherSuccessArgs() { + return Arrays.asList( + Arguments.of( + "basic mono", + new Object[] {2}, + 1, + (Supplier) () -> Mono.just(1).map(Worker::addOne)), + Arguments.of( + "two operations mono", + new Object[] {4}, + 2, + (Supplier) () -> Mono.just(2).map(Worker::addOne).map(Worker::addOne)), + Arguments.of( + "delayed mono", + new Object[] {4}, + 1, + (Supplier) + () -> Mono.just(3).delayElement(Duration.ofMillis(100)).map(Worker::addOne)), + Arguments.of( + "delayed twice mono", + new Object[] {6}, + 2, + (Supplier) + () -> + Mono.just(4) + .delayElement(Duration.ofMillis(100)) + .map(Worker::addOne) + .delayElement(Duration.ofMillis(100)) + .map(Worker::addOne)), + Arguments.of( + "basic flux", + new Object[] {6, 7}, + 2, + (Supplier) () -> Flux.fromIterable(Arrays.asList(5, 6)).map(Worker::addOne)), + Arguments.of( + "two operations flux", + new Object[] {8, 9}, + 4, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(6, 7)).map(Worker::addOne).map(Worker::addOne)), + Arguments.of( + "delayed flux", + new Object[] {8, 9}, + 2, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(7, 8)) + .delayElements(Duration.ofMillis(100)) + .map(Worker::addOne)), + Arguments.of( + "delayed twice flux", + new Object[] {10, 11}, + 4, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(8, 9)) + .delayElements(Duration.ofMillis(100)) + .map(Worker::addOne) + .delayElements(Duration.ofMillis(100)) + .map(Worker::addOne)), + Arguments.of( + "mono from callable", + new Object[] {12}, + 2, + (Supplier) + () -> Mono.fromCallable(() -> Worker.addOne(10)).map(Worker::addOne))); + } + + @ParameterizedTest(name = "Publisher ''{0}'' test") + @MethodSource("publisherSuccessArgs") + void publisherSuccess(String name, Object[] expected, int workSpans, Supplier supplier) { + Object result = Worker.assemblePublisherUnderTrace(supplier); + + if (expected.length == 1) { + assertEquals(expected[0], result); + } else { + assertArrayEquals(expected, (Object[]) result); + } + + SpanMatcher[] matchers = new SpanMatcher[workSpans + 2]; + matchers[0] = + span() + .root() + .operationName("trace-parent") + .resourceName("trace-parent") + .tags(componentTrace(), defaultTags()); + matchers[1] = + span() + .id(Worker.publisherParentId) + .childOf(Worker.traceParentId) + .operationName("publisher-parent") + .resourceName("publisher-parent") + .tags(defaultTags()); + for (int i = 0; i < workSpans; i++) { + matchers[2 + i] = + span() + .childOf(Worker.publisherParentId) + .operationName("addOne") + .resourceName("addOne") + .tags(componentTrace(), defaultTags()); + } + + assertTraces(trace(SORT_BY_START_TIME, matchers)); + } + + // --- Publisher error ----------------------------------------------------- + + static List publisherErrorArgs() { + return Arrays.asList( + Arguments.of( + "mono", (Supplier) () -> Mono.error(new RuntimeException(EXCEPTION_MESSAGE))), + Arguments.of( + "flux", (Supplier) () -> Flux.error(new RuntimeException(EXCEPTION_MESSAGE)))); + } + + @ParameterizedTest(name = "Publisher error ''{0}'' test") + @MethodSource("publisherErrorArgs") + void publisherError(String name, Supplier supplier) { + RuntimeException exception = + assertThrows(RuntimeException.class, () -> Worker.assemblePublisherUnderTrace(supplier)); + assertEquals(EXCEPTION_MESSAGE, exception.getMessage()); + + // It's important that we don't attach errors at the Reactor level so that we don't + // impact the spans on reactor integrations such as netty and lettuce, as reactor is + // more of a context propagation mechanism than something we would be tracking for + // errors — this is ok. + assertTraces( + trace( + SORT_BY_START_TIME, + span() + .root() + .operationName("trace-parent") + .resourceName("trace-parent") + .error() + .tags( + componentTrace(), + error(RuntimeException.class, EXCEPTION_MESSAGE), + defaultTags()), + span() + .id(Worker.publisherParentId) + .childOf(Worker.traceParentId) + .operationName("publisher-parent") + .resourceName("publisher-parent") + .tags(defaultTags()))); + } + + // --- Publisher step error ------------------------------------------------ + + static List publisherStepErrorArgs() { + return Arrays.asList( + Arguments.of( + "basic mono failure", + 1, + (Supplier) + () -> Mono.just(1).map(Worker::addOne).map(i -> Worker.throwException())), + Arguments.of( + "basic flux failure", + 1, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(5, 6)) + .map(Worker::addOne) + .map(i -> Worker.throwException()))); + } + + @ParameterizedTest(name = "Publisher step ''{0}'' test") + @MethodSource("publisherStepErrorArgs") + void publisherStepError(String name, int workSpans, Supplier supplier) { + RuntimeException exception = + assertThrows(RuntimeException.class, () -> Worker.assemblePublisherUnderTrace(supplier)); + assertEquals(EXCEPTION_MESSAGE, exception.getMessage()); + + SpanMatcher[] matchers = new SpanMatcher[workSpans + 2]; + matchers[0] = + span() + .root() + .operationName("trace-parent") + .resourceName("trace-parent") + .error() + .tags( + componentTrace(), error(RuntimeException.class, EXCEPTION_MESSAGE), defaultTags()); + matchers[1] = + span() + .id(Worker.publisherParentId) + .childOf(Worker.traceParentId) + .operationName("publisher-parent") + .resourceName("publisher-parent") + .tags(defaultTags()); + for (int i = 0; i < workSpans; i++) { + matchers[2 + i] = + span() + .childOf(Worker.publisherParentId) + .operationName("addOne") + .resourceName("addOne") + .tags(componentTrace(), defaultTags()); + } + + assertTraces(trace(SORT_BY_START_TIME, matchers)); + } + + // --- Cancel -------------------------------------------------------------- + + static List cancelArgs() { + return Arrays.asList( + Arguments.of("basic mono", (Supplier) () -> Mono.just(1)), + Arguments.of( + "basic flux", (Supplier) () -> Flux.fromIterable(Arrays.asList(5, 6)))); + } + + @ParameterizedTest(name = "Publisher ''{0}'' cancel") + @MethodSource("cancelArgs") + void cancel(String name, Supplier supplier) { + Worker.cancelUnderTrace(supplier); + + assertTraces( + trace( + SORT_BY_START_TIME, + span() + .root() + .operationName("trace-parent") + .resourceName("trace-parent") + .tags(componentTrace(), defaultTags()), + span() + .id(Worker.publisherParentId) + .childOf(Worker.traceParentId) + .operationName("publisher-parent") + .resourceName("publisher-parent") + .tags(defaultTags()))); + } + + // --- Chain spans correct parent ------------------------------------------ + + static List chainParentArgs() { + return Arrays.asList( + Arguments.of( + "basic mono", + 3, + (Supplier) + () -> + Mono.just(1) + .map(Worker::addOne) + .map(Worker::addOne) + .then(Mono.just(1).map(Worker::addOne))), + Arguments.of( + "basic flux", + 5, + (Supplier) + () -> + Flux.fromIterable(Arrays.asList(5, 6)) + .map(Worker::addOne) + .map(Worker::addOne) + .then(Mono.just(1).map(Worker::addOne)))); + } + + @ParameterizedTest(name = "Publisher chain spans have the correct parent for ''{0}''") + @MethodSource("chainParentArgs") + void chainParent(String name, int workSpans, Supplier supplier) { + Worker.assemblePublisherUnderTrace(supplier); + + SpanMatcher[] matchers = new SpanMatcher[workSpans + 2]; + matchers[0] = + span() + .root() + .operationName("trace-parent") + .resourceName("trace-parent") + .tags(componentTrace(), defaultTags()); + matchers[1] = + span() + .id(Worker.publisherParentId) + .childOf(Worker.traceParentId) + .operationName("publisher-parent") + .resourceName("publisher-parent") + .tags(defaultTags()); + for (int i = 0; i < workSpans; i++) { + matchers[2 + i] = + span() + .childOf(Worker.publisherParentId) + .operationName("addOne") + .resourceName("addOne") + .tags(componentTrace(), defaultTags()); + } + + assertTraces(trace(SORT_BY_START_TIME, matchers)); + } + + // --- Correct parents from subscription time (block) ---------------------- + + @Test + void correctParentsFromSubscriptionTimeBlock() { + Mono mono = Mono.just(42).map(Worker::addOne).map(Worker::addTwo); + + Worker.runUnderTraceParent( + () -> { + mono.block(); + return null; + }); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("trace-parent").resourceName("trace-parent"), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()), + span() + .childOf(Worker.traceParentId) + .operationName("addTwo") + .tags(componentTrace(), defaultTags()))); + } + + // --- Correct parents from subscription time (intermediate span) ---------- + + static List subscriptionTimeIntermediateArgs() { + return Arrays.asList( + Arguments.of("basic mono", 1, (Supplier) () -> Mono.just(1).map(Worker::addOne)), + Arguments.of( + "basic flux", + 2, + (Supplier) () -> Flux.fromIterable(Arrays.asList(1, 2)).map(Worker::addOne))); + } + + @ParameterizedTest( + name = "Publisher chain spans have the correct parents from subscription time ''{0}''") + @MethodSource("subscriptionTimeIntermediateArgs") + @SuppressWarnings("unchecked") + void correctParentsFromSubscriptionTime(String name, int workItems, Supplier supplier) { + Worker.assemblePublisherUnderTrace( + () -> { + Object publisher = supplier.get(); + + AgentSpan intermediate = startSpan("test", "intermediate"); + Worker.intermediateId = intermediate.getSpanId(); + AgentScope scope = activateSpan(intermediate); + try { + if (publisher instanceof Mono) { + return ((Mono) publisher).map(Worker::addTwo); + } else if (publisher instanceof Flux) { + return ((Flux) publisher).map(Worker::addTwo); + } + throw new IllegalStateException("Unknown publisher type"); + } finally { + intermediate.finish(); + scope.close(); + } + }); + + SpanMatcher[] matchers = new SpanMatcher[3 + 2 * workItems]; + matchers[0] = + span() + .root() + .operationName("trace-parent") + .resourceName("trace-parent") + .tags(componentTrace(), defaultTags()); + matchers[1] = + span() + .id(Worker.publisherParentId) + .childOf(Worker.traceParentId) + .operationName("publisher-parent") + .resourceName("publisher-parent") + .tags(defaultTags()); + matchers[2] = + span() + .id(Worker.intermediateId) + .childOf(Worker.publisherParentId) + .operationName("intermediate") + .resourceName("intermediate") + .tags(defaultTags()); + for (int i = 0; i < 2 * workItems; i += 2) { + matchers[3 + i] = + span() + .childOf(Worker.publisherParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()); + matchers[4 + i] = + span() + .childOf(Worker.publisherParentId) + .operationName("addTwo") + .tags(componentTrace(), defaultTags()); + } + + assertTraces(trace(SORT_BY_START_TIME, matchers)); + } + + // --- Schedulers ---------------------------------------------------------- + + static List schedulerArgs() { + return Arrays.asList( + Arguments.of("parallel", Schedulers.parallel()), + Arguments.of("single", Schedulers.single()), + Arguments.of("immediate", Schedulers.immediate())); + } + + @ParameterizedTest(name = "Fluxes produce the right number of results on ''{0}'' scheduler") + @MethodSource("schedulerArgs") + void schedulers(String schedulerName, Object scheduler) { + List values = + Flux.fromIterable(Arrays.asList(1, 2, 3, 4)) + .parallel() + .runOn((reactor.core.scheduler.Scheduler) scheduler) + .flatMap(num -> Mono.just(num.toString() + " on " + Thread.currentThread().getName())) + .sequential() + .collectList() + .block(); + + assertEquals(4, values.size()); + } + + // --- Cross-thread context propagation ------------------------------------ + + static List crossThreadArgs() { + return Arrays.asList( + Arguments.of( + "publishOn", + (Supplier) + () -> Flux.just(1, 2).publishOn(Schedulers.parallel()).map(Worker::addOne)), + Arguments.of( + "subscribeOn", + (Supplier) + () -> Flux.just(1, 2).subscribeOn(Schedulers.single()).map(Worker::addOne)), + Arguments.of( + "subscribeOn+publishOn", + (Supplier) + () -> + Flux.just(1, 2) + .subscribeOn(Schedulers.single()) + .publishOn(Schedulers.parallel()) + .map(Worker::addOne))); + } + + @ParameterizedTest(name = "subscribe-time context propagates across threads with ''{0}''") + @MethodSource("crossThreadArgs") + @SuppressWarnings("unchecked") + void crossThreadContextPropagation(String name, Supplier pipeline) { + Worker.runUnderTraceParent( + () -> { + ((Flux) pipeline.get()).collectList().block(); + return null; + }); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("trace-parent").resourceName("trace-parent"), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()))); + } + + // --- No spurious traces outside active trace -------------------------------- + + @Test + void noSpuriousTracesWhenAssembledOutsideTrace() { + Mono.just(1).map(i -> i + 1).block(); + tracer.flush(); + assertEquals( + 0, + writer.getTraceCount(), + () -> "Unexpected traces emitted without active trace: " + writer); + } + + // --- Mono lifecycle spans with parent ------------------------------------ + + @Test + void monoLifecycleSpansWithParent() { + Worker.runUnderTraceParent( + () -> { + Mono.fromCallable(() -> Worker.addOne(0)).doOnNext(v -> Worker.addOne(v)).block(); + return null; + }); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("trace-parent").resourceName("trace-parent"), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()))); + } + + // --- Mono then() --------------------------------------------------------- + + @Test + void monoThenPropagatesContext() { + Worker.runUnderTraceParent( + () -> { + Mono.create( + sink -> { + Worker.addOne(0); + sink.success(); + }) + .then( + Mono.create( + sink -> { + Worker.addOne(0); + sink.success(); + })) + .block(); + return null; + }); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("trace-parent").resourceName("trace-parent"), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()), + span() + .childOf(Worker.traceParentId) + .operationName("addOne") + .tags(componentTrace(), defaultTags()))); + } + + // --- windowUntil does not throw NPE on advice ---------------------------- + + @Test + void windowUntilDoesNotThrowNpe() { + Long count = Flux.range(1, 100).windowUntil(i -> i % 10 == 0).count().block(); + assertEquals(11, count); + } +} diff --git a/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/SubscriptionTest.java b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/SubscriptionTest.java new file mode 100644 index 00000000000..1f6d091df1a --- /dev/null +++ b/dd-java-agent/instrumentation/reactor-core-3.1/src/test/java/testdog/trace/instrumentation/reactor/core/SubscriptionTest.java @@ -0,0 +1,324 @@ +package testdog.trace.instrumentation.reactor.core; + +import static datadog.trace.agent.test.assertions.SpanMatcher.span; +import static datadog.trace.agent.test.assertions.TraceMatcher.SORT_BY_START_TIME; +import static datadog.trace.agent.test.assertions.TraceMatcher.trace; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.activateSpan; +import static datadog.trace.bootstrap.instrumentation.api.AgentTracer.startSpan; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.agent.test.assertions.TraceMatcher; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import org.junit.jupiter.api.Test; +import reactor.core.publisher.DirectProcessor; +import reactor.core.publisher.EmitterProcessor; +import reactor.core.publisher.Flux; +import reactor.core.publisher.FluxProcessor; +import reactor.core.publisher.Mono; +import reactor.core.publisher.TopicProcessor; +import reactor.core.publisher.WorkQueueProcessor; + +class SubscriptionTest extends AbstractInstrumentationTest { + + static class Connection { + static int query() { + AgentSpan span = startSpan("test", "Connection.query"); + span.finish(); + return new Random().nextInt(); + } + } + + // --- Processor-based subscription tests ---------------------------------- + + @Test + void directProcessorSingleSubscriberPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(DirectProcessor.create(), 1); + } + + @Test + void emitterProcessorSingleSubscriberPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(EmitterProcessor.create(), 1); + } + + @Test + void topicProcessorSingleSubscriberPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(TopicProcessor.create(), 1); + } + + @Test + void workQueueProcessorSingleSubscriberPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(WorkQueueProcessor.create(), 1); + } + + @Test + void directProcessorMultipleSubscribersPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(DirectProcessor.create(), 3); + } + + @Test + void emitterProcessorMultipleSubscribersPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(EmitterProcessor.create(), 3); + } + + @Test + void topicProcessorMultipleSubscribersPropagatesParentSpan() throws InterruptedException { + verifyProcessorPropagation(TopicProcessor.create(), 3); + } + + @SuppressWarnings("unchecked") + private void verifyProcessorPropagation(Object rawProcessor, int consumers) + throws InterruptedException { + FluxProcessor processor = + (FluxProcessor) rawProcessor; + CountDownLatch published = new CountDownLatch(consumers); + CountDownLatch subscribed = new CountDownLatch(consumers); + + for (int i = 0; i < consumers; i++) { + Thread t = + new Thread( + () -> { + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + processor.subscribe( + connection -> { + Connection.query(); + published.countDown(); + }); + subscribed.countDown(); + try { + published.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } finally { + parent.finish(); + } + }); + t.start(); + } + + subscribed.await(); + processor.sink().next(new Connection()); + published.await(); + + // Each subscriber gets its own trace with parent -> child relationship + TraceMatcher[] traceMatchers = new TraceMatcher[consumers]; + for (int i = 0; i < consumers; i++) { + traceMatchers[i] = + trace( + SORT_BY_START_TIME, + span().root().operationName("parent"), + span().childOfPrevious().operationName("Connection.query")); + } + assertTraces(traceMatchers); + } + + // --- Broadcasting flux --------------------------------------------------- + + @Test + void broadcastingFluxPropagatesParentSpan() throws InterruptedException { + Flux connection = Flux.just(new Connection()).publish().autoConnect(); + + CountDownLatch published = new CountDownLatch(1); + CountDownLatch subscribed = new CountDownLatch(1); + + Thread t = + new Thread( + () -> { + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + connection.subscribe( + c -> { + Connection.query(); + published.countDown(); + }); + subscribed.countDown(); + try { + published.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } finally { + parent.finish(); + } + }); + t.start(); + subscribed.await(); + published.await(); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("parent"), + span().childOfPrevious().operationName("Connection.query"))); + } + + // --- Mono subscription propagates parent span ---------------------------- + + @Test + void monoSubscriptionPropagatesParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Mono connection = Mono.create(emitter -> emitter.success(new Connection())); + connection.subscribe( + c -> { + Connection.query(); + latch.countDown(); + }); + } finally { + parent.finish(); + } + latch.await(); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("parent"), + span().childOfPrevious().operationName("Connection.query"))); + } + + // --- Flux subscription propagates parent span ---------------------------- + + @Test + void fluxSubscriptionPropagatesParentSpan() throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Flux connection = + Flux.create( + emitter -> { + emitter.next(new Connection()); + emitter.complete(); + }); + connection.subscribe( + c -> { + Connection.query(); + latch.countDown(); + }); + } finally { + parent.finish(); + } + latch.await(); + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("parent"), + span().childOfPrevious().operationName("Connection.query"))); + } + + // --- Mono then() propagates context across chained publishers ------------ + + @Test + void monoThenPropagatesContext() { + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Mono.create( + sink -> { + AgentSpan child1 = startSpan("test", "child1"); + child1.finish(); + sink.success(); + }) + .then( + Mono.create( + sink -> { + AgentSpan child2 = startSpan("test", "child2"); + child2.finish(); + sink.success(); + })) + .block(); + } finally { + parent.finish(); + } + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("parent"), + span().childOfPrevious().operationName("child1"), + span().childOfIndex(0).operationName("child2"))); + } + + // --- Mono lifecycle spans without parent produce separate traces --------- + + @Test + void monoLifecycleSpansWithoutParentProduceSeparateTraces() { + Mono.fromCallable( + () -> { + AgentSpan span = startSpan("test", "span"); + span.finish(); + return Mono.just("Hello World"); + }) + .doOnNext( + v -> { + AgentSpan onNext = startSpan("test", "onNext"); + onNext.finish(); + }) + .doFinally( + signal -> { + AgentSpan finallySpan = startSpan("test", "finally"); + finallySpan.finish(); + }) + .doAfterTerminate( + () -> { + AgentSpan after = startSpan("test", "after"); + after.finish(); + }) + .block(); + + // Without a parent span, each manually started span is its own trace + assertTraces( + trace(span().root().operationName("span")), + trace(span().root().operationName("onNext")), + trace(span().root().operationName("after")), + trace(span().root().operationName("finally"))); + } + + // --- Mono lifecycle spans with parent all join the parent trace ---------- + + @Test + void monoLifecycleSpansWithParentJoinParentTrace() { + AgentSpan parent = startSpan("test", "parent"); + try (AgentScope scope = activateSpan(parent)) { + Mono.fromCallable( + () -> { + AgentSpan span = startSpan("test", "span"); + span.finish(); + return Mono.just("Hello World"); + }) + .doOnNext( + v -> { + AgentSpan onNext = startSpan("test", "onNext"); + onNext.finish(); + }) + .doFinally( + signal -> { + AgentSpan finallySpan = startSpan("test", "finally"); + finallySpan.finish(); + }) + .doAfterTerminate( + () -> { + AgentSpan after = startSpan("test", "after"); + after.finish(); + }) + .block(); + } finally { + parent.finish(); + } + + assertTraces( + trace( + SORT_BY_START_TIME, + span().root().operationName("parent"), + span().childOfPrevious().operationName("span"), + span().childOfIndex(0).operationName("onNext"), + span().childOfIndex(0).operationName("after"), + span().childOfIndex(0).operationName("finally"))); + } +}