From 58311a96ae495620f403cbef1ff4967607def096 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 27 May 2026 14:33:26 -0400 Subject: [PATCH 1/7] fix(okhttp): capture scope at AsyncCall. to keep traces intact under virtual-thread Dispatcher load OkHttp's Dispatcher recursively promotes queued AsyncCalls from inside finished(), running on a worker thread that has a *different* call's parent scope active. The existing TaskRunnerInstrumentation captures that wrong scope when the promoted call is executed, so under a virtual-thread-per-task Dispatcher (or any executor where the same recursion exposes the issue) okhttp.request spans cross-contaminate between concurrent caller traces. Single-shot requests don't trigger this because nothing is queued. Fix: add AsyncCallInstrumentation that captures the active scope at AsyncCall. (i.e. the user thread that ran enqueue(), where the right parent is active) and stores it in the shared ContextStore. RunnableInstrumentation then activates that state on AsyncCall.run() entry, overriding whatever scope TaskRunner inherited from the dispatcher-recursion path. Covers both class locations (3.x and 4.x). Muzzle passes against 24 OkHttp versions in [3.0.0, 5.3.2]. All 168 pre-existing forked OkHttp tests still pass. Tests added: * okhttp-3.0/src/vthread21Test/.../OkHttpVirtualThreadDispatcherTest: concurrent-burst test (16 parents x 4 requests through a virtual- thread Dispatcher with capacity 4) plus two single-shot baselines. The concurrent-burst test fails on master and v1.61.1 without the fix and passes with it. * java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest: pure-executor regression coverage for Thread.ofVirtual().factory(), the .name(prefix, start) builder variant, recursive submission, and propagation across virtual-thread unmount. Co-Authored-By: Claude Opus 4.7 (1M context) --- ...hreadPerTaskExecutorVirtualThreadTest.java | 190 +++++++++++ .../okhttp/okhttp-3.0/build.gradle | 31 ++ .../okhttp3/AsyncCallInstrumentation.java | 78 +++++ .../OkHttpVirtualThreadDispatcherTest.java | 317 ++++++++++++++++++ 4 files changed, 616 insertions(+) create mode 100644 dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java create mode 100644 dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java create mode 100644 dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java diff --git a/dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java b/dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java new file mode 100644 index 00000000000..53f7a348a44 --- /dev/null +++ b/dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java @@ -0,0 +1,190 @@ +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.api.Trace; +import datadog.trace.core.DDSpan; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import org.junit.jupiter.api.Test; + +/** + * Reproductions for the executor swap profiling-backend PR#8520 made for OkHttp's Dispatcher + * (cached thread pool → {@code Executors.newThreadPerTaskExecutor(Thread.ofVirtual()...)}). + * + *

Each test asserts that an {@code @Trace}-annotated child method invoked from inside a task + * submitted to the virtual-thread-per-task executor attaches to the parent span that was active at + * submission time. If propagation breaks, the child either lands in a different trace or floats + * free as a root span and the {@code assertEquals(parentSpanId, childSpan.getParentId())} check + * fails. + * + *

This mirrors what would happen with OkHttp: the {@code okhttp} client span is created by + * {@code TracingInterceptor.intercept()} on the executor thread using {@code activeSpan()}, so "is + * the parent scope active on the worker?" is the entire question. + */ +class ThreadPerTaskExecutorVirtualThreadTest extends AbstractInstrumentationTest { + + @Trace(operationName = "parent") + void underParentTrace(int expectedChildren, Runnable body) { + body.run(); + blockUntilChildSpansFinished(expectedChildren); + } + + @Trace(operationName = "child") + static void child() {} + + @Test + void virtualThreadFactory_propagatesActiveScope() throws Exception { + ExecutorService executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()); + try { + underParentTrace(1, () -> executor.execute(ThreadPerTaskExecutorVirtualThreadTest::child)); + } finally { + executor.shutdown(); + } + + writer.waitForTraces(1); + List trace = writer.get(0); + assertEquals(2, trace.size(), "expected parent + child span"); + DDSpan parentSpan = findByOp(trace, "parent"); + DDSpan childSpan = findByOp(trace, "child"); + assertNotNull(parentSpan, "parent span should exist"); + assertNotNull(childSpan, "child span should exist"); + assertEquals( + parentSpan.getSpanId(), + childSpan.getParentId(), + "child span should be parented under the active span at executor.execute()"); + } + + @Test + void namedVirtualThreadFactory_propagatesActiveScope() throws Exception { + // Exact builder shape from profiling-backend PR#8520: + // Thread.ofVirtual().name("okhttp-" + track + "-", 0).factory() + ExecutorService executor = + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-test-", 0).factory()); + try { + underParentTrace(1, () -> executor.execute(ThreadPerTaskExecutorVirtualThreadTest::child)); + } finally { + executor.shutdown(); + } + + writer.waitForTraces(1); + List trace = writer.get(0); + DDSpan parentSpan = findByOp(trace, "parent"); + DDSpan childSpan = findByOp(trace, "child"); + assertNotNull(parentSpan, "parent span should exist"); + assertNotNull(childSpan, "child span should exist"); + assertEquals( + parentSpan.getSpanId(), + childSpan.getParentId(), + "the .name(prefix, start) builder should not break propagation"); + } + + /** + * Mirrors OkHttp's dispatcher recursion: a task running on the executor submits more work back to + * the same executor. In OkHttp this is {@code Dispatcher.finished() → promoteAndExecute() + * → executorService.execute(nextAsyncCall)}, called from inside an {@code AsyncCall.run()} + * on a dispatcher thread. + * + *

Both child spans should land in the parent's trace. If the worker thread loses the parent + * scope between the outer activation and the inner submission, the second child either becomes a + * new root span or gets attached to the wrong sibling. + */ + @Test + void recursiveSubmissionFromWorkerThread_keepsTraceConnected() throws Exception { + ExecutorService executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()); + try { + underParentTrace( + 2, + () -> + executor.execute( + () -> { + child(); + executor.execute(ThreadPerTaskExecutorVirtualThreadTest::child); + })); + } finally { + executor.shutdown(); + } + + writer.waitForTraces(1); + List trace = writer.get(0); + DDSpan parentSpan = findByOp(trace, "parent"); + assertNotNull(parentSpan, "parent span should exist"); + long parentTraceId = parentSpan.getTraceId().toLong(); + long parentSpanId = parentSpan.getSpanId(); + + long childCount = + trace.stream().filter(s -> "child".contentEquals(s.getOperationName())).count(); + assertEquals(2, childCount, "both child spans should land in the same trace as the parent"); + + trace.stream() + .filter(s -> "child".contentEquals(s.getOperationName())) + .forEach( + s -> { + assertEquals( + parentTraceId, + s.getTraceId().toLong(), + "child span must share the parent's trace"); + assertEquals( + parentSpanId, + s.getParentId(), + "child span must attach to the parent, not float free"); + }); + } + + /** + * Forces the virtual thread to unmount and remount mid-task by sleeping between two child spans. + * This is the case OkHttp actually hits in practice — every blocking socket read unmounts + * the carrier, and the scope stack has to be reinstated by {@code VirtualThreadInstrumentation} + * on each remount. The fix log for this area is long (#10931, #11009, #11111), so it is worth + * pinning down with a regression test. + */ + @Test + void virtualThreadFactory_propagatesAcrossUnmount() throws Exception { + ExecutorService executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()); + try { + underParentTrace( + 2, + () -> + executor.execute( + () -> { + child(); + try { + // Sleep is a JDK 21+ carrier-unmounting parking point for virtual threads. + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + child(); + })); + } finally { + executor.shutdown(); + } + + writer.waitForTraces(1); + List trace = writer.get(0); + DDSpan parentSpan = findByOp(trace, "parent"); + assertNotNull(parentSpan, "parent span should exist"); + long parentSpanId = parentSpan.getSpanId(); + + long childCount = + trace.stream().filter(s -> "child".contentEquals(s.getOperationName())).count(); + assertEquals(2, childCount, "both child spans should be captured"); + + trace.stream() + .filter(s -> "child".contentEquals(s.getOperationName())) + .forEach( + s -> + assertEquals( + parentSpanId, + s.getParentId(), + "child span before and after virtual-thread unmount must both attach to parent")); + } + + private static DDSpan findByOp(List spans, String op) { + return spans.stream() + .filter(s -> op.contentEquals(s.getOperationName())) + .findFirst() + .orElse(null); + } +} diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle index 5ce7c997f99..e2ce1db8684 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle @@ -8,9 +8,30 @@ muzzle { } apply from: "$rootDir/gradle/java.gradle" +// Use slf4j-simple for tests; logback's synchronized appenders can pin virtual-thread carriers +// when many vthreads log concurrently, which deadlocks this module's vthread21Test suite. +apply from: "$rootDir/gradle/slf4j-simple.gradle" addTestSuiteForDir('latestDepTest', 'test') +// Separate suite for tests that need the JDK 21+ virtual-thread API. Kept apart from the +// default test suite so that the existing OkHttp 3.0 tests keep their Java 1.8 baseline. +addTestSuite('vthread21Test') + +tasks.named("compileVthread21TestJava", JavaCompile) { + configureCompiler(it, 21) +} + +tasks.named("vthread21Test", Test) { + javaLauncher = javaToolchains.launcherFor { + languageVersion = JavaLanguageVersion.of(21) + } +} + +tasks.named("check") { + dependsOn "vthread21Test" +} + dependencies { compileOnly(group: 'com.squareup.okhttp3', name: 'okhttp', version: '3.0.0') @@ -38,4 +59,14 @@ dependencies { testRuntimeOnly(project(':dd-java-agent:instrumentation:datadog:asm:iast-instrumenter')) testRuntimeOnly(project(':dd-java-agent:instrumentation:java:java-net:java-net-1.8')) + + // vthread21Test inherits testImplementation via addTestSuite's extendsFrom wiring. + vthread21TestImplementation(group: 'com.squareup.okhttp3', name: 'okhttp', version: '3.11.0') + vthread21TestImplementation(group: 'com.squareup.okio', name: 'okio', version: '1.14.0') + + // Pull in the JDK-21+ concurrent / lang instrumentations so the test installs the same + // TaskRunnerInstrumentation + VirtualThreadInstrumentation chain that profiling-backend + // exercises in production. + vthread21TestRuntimeOnly project(':dd-java-agent:instrumentation:java:java-concurrent:java-concurrent-21.0') + vthread21TestRuntimeOnly project(':dd-java-agent:instrumentation:java:java-lang:java-lang-21.0') } diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java new file mode 100644 index 00000000000..0d9d9295b99 --- /dev/null +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java @@ -0,0 +1,78 @@ +package datadog.trace.instrumentation.okhttp3; + +import static datadog.trace.bootstrap.instrumentation.java.concurrent.AdviceUtils.capture; +import static java.util.Collections.singletonMap; +import static net.bytebuddy.matcher.ElementMatchers.isConstructor; + +import com.google.auto.service.AutoService; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.agent.tooling.InstrumenterModule; +import datadog.trace.bootstrap.InstrumentationContext; +import datadog.trace.bootstrap.instrumentation.java.concurrent.State; +import java.util.Map; +import net.bytebuddy.asm.Advice; + +/** + * Captures the active scope at {@code AsyncCall.} (i.e. the moment {@code + * Call.enqueue(callback)} was invoked on the user's thread) and stores it in the {@code + * ContextStore} shared with the rest of the concurrent instrumentation. {@code + * RunnableInstrumentation} then re-activates that scope on {@code AsyncCall.run()} entry, which + * overrides whatever scope {@code TaskRunner.run()} (or {@code beforeExecute}) put in place from + * the dispatcher's worker thread. + * + *

Without this, {@code TaskRunnerInstrumentation} captures whatever scope happens to be active + * on the worker thread when {@code Dispatcher.promoteAndExecute()} dequeues and submits the call — + * and when promotion runs from inside {@code Dispatcher.finished()} (i.e. recursively from a + * different AsyncCall's run()), that scope belongs to the finishing call, not to the + * caller who actually enqueued this AsyncCall. Result: under concurrent OkHttp load, {@code + * okhttp.request} spans cross-contaminate between traces. + * + *

The class moved between OkHttp 3.x and 4.x: + * + *

    + *
  • OkHttp 3.x — {@code okhttp3.RealCall$AsyncCall} + *
  • OkHttp 4.x — {@code okhttp3.internal.connection.RealCall$AsyncCall} + *
+ * + * Both are inner classes of {@code RealCall} and both transitively implement {@link Runnable}. + */ +@AutoService(InstrumenterModule.class) +public final class AsyncCallInstrumentation extends InstrumenterModule.Tracing + implements Instrumenter.ForKnownTypes, Instrumenter.HasMethodAdvice { + + public AsyncCallInstrumentation() { + // Re-use the existing "okhttp" / "okhttp-3" instrumentation names so we don't introduce a + // separately-toggleable feature flag (DD_TRACE_OKHTTP_ASYNC_CALL_ENABLED). The capture here + // is conceptually part of the OkHttp instrumentation — if you disable OkHttp tracing, you + // also disable this capture, which is the right behavior. + super("okhttp", "okhttp-3"); + } + + @Override + public String[] knownMatchingTypes() { + return new String[] { + "okhttp3.RealCall$AsyncCall", // OkHttp 3.x + "okhttp3.internal.connection.RealCall$AsyncCall", // OkHttp 4.x + }; + } + + @Override + public Map contextStore() { + // Same Runnable -> State store that RunnableInstrumentation reads from. + return singletonMap("java.lang.Runnable", State.class.getName()); + } + + @Override + public void methodAdvice(MethodTransformer transformer) { + transformer.applyAdvice(isConstructor(), getClass().getName() + "$Construct"); + } + + public static final class Construct { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void captureScope(@Advice.This Runnable asyncCall) { + // AdviceUtils.capture is a no-op when async propagation is disabled or there's no active + // span — same behavior as the rest of the concurrent instrumentation. + capture(InstrumentationContext.get(Runnable.class, State.class), asyncCall); + } + } +} diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java new file mode 100644 index 00000000000..6def8c6ec33 --- /dev/null +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java @@ -0,0 +1,317 @@ +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.sun.net.httpserver.HttpServer; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.AgentTracer; +import datadog.trace.core.DDSpan; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import okhttp3.Call; +import okhttp3.Callback; +import okhttp3.Dispatcher; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * End-to-end reproduction of profiling-backend PR#8520: swap OkHttp's Dispatcher executor from + * {@code Executors.newCachedThreadPool(...)} to {@code + * Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name(prefix, start).factory())}. + * + *

The test runs both shapes against the same {@link HttpServer} mock. Inside a manually + * activated "parent" span it does {@code client.newCall(request).enqueue(callback)} and waits on a + * latch for the callback. The agent's OkHttp instrumentation injects {@code TracingInterceptor}, + * which creates the {@code okhttp.request} client span using whatever scope is active on the + * dispatcher worker. The assertions verify the client span lands under the parent — i.e., the + * dispatcher's worker thread saw the propagated scope. + * + *

If propagation fails for the virtual-thread shape (the failure profiling-backend is reporting) + * the {@code okhttp.request} span will either become a root span in its own trace or be parented + * under nothing, and {@code assertEquals(parentSpan.getSpanId(), okhttpSpan.getParentId())} fails. + */ +class OkHttpVirtualThreadDispatcherTest extends AbstractInstrumentationTest { + + private static HttpServer mockServer; + private static String baseUrl; + + @BeforeAll + static void startServer() throws IOException { + mockServer = HttpServer.create(new InetSocketAddress("localhost", 0), 64); + mockServer.createContext( + "/ok", + exchange -> { + byte[] body = "ok".getBytes(); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + // Default executor is single-threaded — give the server real concurrency so the test isn't + // bottlenecked on the mock backend itself. + mockServer.setExecutor(Executors.newCachedThreadPool()); + mockServer.start(); + baseUrl = + "http://" + + mockServer.getAddress().getHostString() + + ":" + + mockServer.getAddress().getPort(); + } + + @AfterAll + static void stopServer() { + if (mockServer != null) { + mockServer.stop(0); + mockServer = null; + } + } + + /** Activate a manual parent span, run the OkHttp call, wait for the okhttp.request child. */ + private void runUnderParent(OkHttpClient client, CountDownLatch done) { + AgentSpan parentSpan = AgentTracer.startSpan("test", "parent"); + try (AgentScope ignored = AgentTracer.activateSpan(parentSpan)) { + Request request = new Request.Builder().url(baseUrl + "/ok").build(); + client + .newCall(request) + .enqueue( + new Callback() { + @Override + public void onResponse(Call call, Response response) throws IOException { + response.body().close(); + done.countDown(); + } + + @Override + public void onFailure(Call call, IOException e) { + done.countDown(); + } + }); + if (!done.await(10, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting for OkHttp callback"); + } + // Wait for the okhttp.request child to finish before closing the parent scope so the trace + // collector sees a complete trace. + blockUntilChildSpansFinished(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } finally { + parentSpan.finish(); + } + } + + @Test + void cachedThreadPoolDispatcher_parentsOkHttpSpanUnderParent() throws Exception { + ExecutorService dispatcherExecutor = Executors.newCachedThreadPool(); + OkHttpClient client = buildClient(dispatcherExecutor); + try { + runUnderParent(client, new CountDownLatch(1)); + } finally { + dispatcherExecutor.shutdown(); + } + + assertOkHttpSpanParentedUnderParent(); + } + + @Test + void virtualThreadPerTaskDispatcher_parentsOkHttpSpanUnderParent() throws Exception { + // Exact shape from profiling-backend PR#8520. + ExecutorService dispatcherExecutor = + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-test-", 0).factory()); + OkHttpClient client = buildClient(dispatcherExecutor); + try { + runUnderParent(client, new CountDownLatch(1)); + } finally { + dispatcherExecutor.shutdown(); + } + + assertOkHttpSpanParentedUnderParent(); + } + + /** + * Concurrent stress test: spin up N independent parent traces, each enqueueing M OkHttp requests + * through the same virtual-thread dispatcher. Dispatcher capacity is intentionally set below N*M + * so that some calls get queued and then promoted from {@code Dispatcher.finished()} running on a + * dispatcher worker thread (a different parent's worker). + * + *

If the agent captures the worker's currently-active scope when the promoted call is + * submitted — instead of the scope active where the original {@code enqueue()} happened + * — the {@code okhttp.request} span will land in the wrong parent's trace. This test fails + * loudly on that cross-contamination. + */ + @Test + void concurrentVirtualThreadPerTaskDispatcher_keepsEachTraceSeparate() throws Exception { + int parentCount = 16; + int requestsPerParent = 4; + int totalRequests = parentCount * requestsPerParent; + + ExecutorService dispatcherExecutor = + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-burst-", 0).factory()); + OkHttpClient client = buildClient(dispatcherExecutor); + // Force queue contention: capacity is much smaller than the total in-flight requests, so + // many calls will be promoted from finished() rather than direct from enqueue(). + Dispatcher dispatcher = client.dispatcher(); + dispatcher.setMaxRequests(4); + dispatcher.setMaxRequestsPerHost(4); + + ExecutorService parentRunner = Executors.newFixedThreadPool(parentCount); + CountDownLatch allParentsDone = new CountDownLatch(parentCount); + AtomicReference failure = new AtomicReference<>(); + + try { + for (int i = 0; i < parentCount; i++) { + parentRunner.submit( + () -> { + try { + runParentBurst(client, requestsPerParent); + } catch (Throwable t) { + failure.compareAndSet(null, t); + } finally { + allParentsDone.countDown(); + } + }); + } + assertTrue(allParentsDone.await(60, TimeUnit.SECONDS), "parent threads timed out"); + } finally { + parentRunner.shutdown(); + dispatcherExecutor.shutdown(); + } + + if (failure.get() != null) { + throw new AssertionError("a parent-runner thread threw", failure.get()); + } + + // Each parent produces one trace containing 1 parent span + M okhttp.request spans. + writer.waitForTraces(parentCount); + + // Map trace-id -> spans-in-that-trace and assert structure. + Map> tracesByRoot = new HashMap<>(); + for (List trace : writer) { + DDSpan parentSpan = findByOp(trace, "parent"); + assertNotNull(parentSpan, "every collected trace should have a parent span"); + tracesByRoot.put(parentSpan.getSpanId(), trace); + } + assertEquals( + parentCount, + tracesByRoot.size(), + "expected one distinct trace per parent burst (no cross-trace contamination)"); + + int totalOkhttpSpans = 0; + List contamination = new ArrayList<>(); + for (Map.Entry> entry : tracesByRoot.entrySet()) { + long parentSpanId = entry.getKey(); + List trace = entry.getValue(); + int okhttpCountInThisTrace = 0; + for (DDSpan span : trace) { + if ("okhttp.request".contentEquals(span.getOperationName())) { + okhttpCountInThisTrace++; + if (span.getParentId() != parentSpanId) { + contamination.add( + "okhttp.request span " + + span.getSpanId() + + " has parentId=" + + span.getParentId() + + " but it lives in the trace rooted at " + + parentSpanId); + } + } + } + totalOkhttpSpans += okhttpCountInThisTrace; + assertEquals( + requestsPerParent, + okhttpCountInThisTrace, + "trace rooted at parent " + parentSpanId + " has wrong child count"); + } + + assertEquals(totalRequests, totalOkhttpSpans, "total okhttp.request spans across all traces"); + assertTrue( + contamination.isEmpty(), + "found cross-parented okhttp.request spans:\n - " + String.join("\n - ", contamination)); + } + + /** + * One parent burst: M OkHttp requests under a single freshly-started parent span. Deliberately + * does not block on per-parent child-span accounting — the whole point of the test + * is to detect when children leak to a sibling's trace, and per-parent blocking would just turn + * that into a timeout instead of producing a useful assertion message. Wait for the HTTP + * callbacks (so the request actually ran), then close the parent. + */ + private void runParentBurst(OkHttpClient client, int requestsPerParent) { + AgentSpan parentSpan = AgentTracer.startSpan("test", "parent"); + try (AgentScope ignored = AgentTracer.activateSpan(parentSpan)) { + CountDownLatch done = new CountDownLatch(requestsPerParent); + for (int i = 0; i < requestsPerParent; i++) { + Request request = new Request.Builder().url(baseUrl + "/ok").build(); + client + .newCall(request) + .enqueue( + new Callback() { + @Override + public void onResponse(Call call, Response response) throws IOException { + response.body().close(); + done.countDown(); + } + + @Override + public void onFailure(Call call, IOException e) { + done.countDown(); + } + }); + } + if (!done.await(30, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting for OkHttp callbacks"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } finally { + parentSpan.finish(); + } + } + + private static OkHttpClient buildClient(ExecutorService dispatcherExecutor) { + Dispatcher dispatcher = new Dispatcher(dispatcherExecutor); + return new OkHttpClient.Builder().dispatcher(dispatcher).build(); + } + + private void assertOkHttpSpanParentedUnderParent() throws Exception { + writer.waitForTraces(1); + List trace = writer.get(0); + DDSpan parentSpan = findByOp(trace, "parent"); + DDSpan okhttpSpan = findByOp(trace, "okhttp.request"); + assertNotNull(parentSpan, "parent span should exist"); + assertNotNull( + okhttpSpan, + "okhttp.request client span should exist; if missing, propagation may have produced an" + + " orphan trace instead"); + assertEquals( + parentSpan.getTraceId().toLong(), + okhttpSpan.getTraceId().toLong(), + "okhttp.request span must share the parent's trace id"); + assertEquals( + parentSpan.getSpanId(), + okhttpSpan.getParentId(), + "okhttp.request span must be parented under the parent span active at enqueue() time"); + } + + private static DDSpan findByOp(List spans, String op) { + return spans.stream() + .filter(s -> op.contentEquals(s.getOperationName())) + .findFirst() + .orElse(null); + } +} From 84979cad0a95ba74b1f14cb2e2aceb4ec2a0d556 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Fri, 29 May 2026 07:37:13 -0400 Subject: [PATCH 2/7] fix(okhttp): address review feedback on virtual-thread dispatcher test suite - Replace javaLauncher override with testJvmConstraints so lower-JDK shards skip vthread21Test rather than forcing JDK 21 - Drop OkHttp 4.x from AsyncCallInstrumentation.knownMatchingTypes() until test coverage exists; trim stale 4.x javadoc - Remove ThreadPerTaskExecutorVirtualThreadTest (no OkHttp dependency, belongs in a separate PR) Co-Authored-By: Claude Sonnet 4.6 --- ...hreadPerTaskExecutorVirtualThreadTest.java | 190 ------------------ .../okhttp/okhttp-3.0/build.gradle | 4 +- .../okhttp3/AsyncCallInstrumentation.java | 13 +- 3 files changed, 5 insertions(+), 202 deletions(-) delete mode 100644 dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java diff --git a/dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java b/dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java deleted file mode 100644 index 53f7a348a44..00000000000 --- a/dd-java-agent/instrumentation/java/java-concurrent/java-concurrent-21.0/src/test/java/ThreadPerTaskExecutorVirtualThreadTest.java +++ /dev/null @@ -1,190 +0,0 @@ -import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertNotNull; - -import datadog.trace.agent.test.AbstractInstrumentationTest; -import datadog.trace.api.Trace; -import datadog.trace.core.DDSpan; -import java.util.List; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; -import org.junit.jupiter.api.Test; - -/** - * Reproductions for the executor swap profiling-backend PR#8520 made for OkHttp's Dispatcher - * (cached thread pool → {@code Executors.newThreadPerTaskExecutor(Thread.ofVirtual()...)}). - * - *

Each test asserts that an {@code @Trace}-annotated child method invoked from inside a task - * submitted to the virtual-thread-per-task executor attaches to the parent span that was active at - * submission time. If propagation breaks, the child either lands in a different trace or floats - * free as a root span and the {@code assertEquals(parentSpanId, childSpan.getParentId())} check - * fails. - * - *

This mirrors what would happen with OkHttp: the {@code okhttp} client span is created by - * {@code TracingInterceptor.intercept()} on the executor thread using {@code activeSpan()}, so "is - * the parent scope active on the worker?" is the entire question. - */ -class ThreadPerTaskExecutorVirtualThreadTest extends AbstractInstrumentationTest { - - @Trace(operationName = "parent") - void underParentTrace(int expectedChildren, Runnable body) { - body.run(); - blockUntilChildSpansFinished(expectedChildren); - } - - @Trace(operationName = "child") - static void child() {} - - @Test - void virtualThreadFactory_propagatesActiveScope() throws Exception { - ExecutorService executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()); - try { - underParentTrace(1, () -> executor.execute(ThreadPerTaskExecutorVirtualThreadTest::child)); - } finally { - executor.shutdown(); - } - - writer.waitForTraces(1); - List trace = writer.get(0); - assertEquals(2, trace.size(), "expected parent + child span"); - DDSpan parentSpan = findByOp(trace, "parent"); - DDSpan childSpan = findByOp(trace, "child"); - assertNotNull(parentSpan, "parent span should exist"); - assertNotNull(childSpan, "child span should exist"); - assertEquals( - parentSpan.getSpanId(), - childSpan.getParentId(), - "child span should be parented under the active span at executor.execute()"); - } - - @Test - void namedVirtualThreadFactory_propagatesActiveScope() throws Exception { - // Exact builder shape from profiling-backend PR#8520: - // Thread.ofVirtual().name("okhttp-" + track + "-", 0).factory() - ExecutorService executor = - Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-test-", 0).factory()); - try { - underParentTrace(1, () -> executor.execute(ThreadPerTaskExecutorVirtualThreadTest::child)); - } finally { - executor.shutdown(); - } - - writer.waitForTraces(1); - List trace = writer.get(0); - DDSpan parentSpan = findByOp(trace, "parent"); - DDSpan childSpan = findByOp(trace, "child"); - assertNotNull(parentSpan, "parent span should exist"); - assertNotNull(childSpan, "child span should exist"); - assertEquals( - parentSpan.getSpanId(), - childSpan.getParentId(), - "the .name(prefix, start) builder should not break propagation"); - } - - /** - * Mirrors OkHttp's dispatcher recursion: a task running on the executor submits more work back to - * the same executor. In OkHttp this is {@code Dispatcher.finished() → promoteAndExecute() - * → executorService.execute(nextAsyncCall)}, called from inside an {@code AsyncCall.run()} - * on a dispatcher thread. - * - *

Both child spans should land in the parent's trace. If the worker thread loses the parent - * scope between the outer activation and the inner submission, the second child either becomes a - * new root span or gets attached to the wrong sibling. - */ - @Test - void recursiveSubmissionFromWorkerThread_keepsTraceConnected() throws Exception { - ExecutorService executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()); - try { - underParentTrace( - 2, - () -> - executor.execute( - () -> { - child(); - executor.execute(ThreadPerTaskExecutorVirtualThreadTest::child); - })); - } finally { - executor.shutdown(); - } - - writer.waitForTraces(1); - List trace = writer.get(0); - DDSpan parentSpan = findByOp(trace, "parent"); - assertNotNull(parentSpan, "parent span should exist"); - long parentTraceId = parentSpan.getTraceId().toLong(); - long parentSpanId = parentSpan.getSpanId(); - - long childCount = - trace.stream().filter(s -> "child".contentEquals(s.getOperationName())).count(); - assertEquals(2, childCount, "both child spans should land in the same trace as the parent"); - - trace.stream() - .filter(s -> "child".contentEquals(s.getOperationName())) - .forEach( - s -> { - assertEquals( - parentTraceId, - s.getTraceId().toLong(), - "child span must share the parent's trace"); - assertEquals( - parentSpanId, - s.getParentId(), - "child span must attach to the parent, not float free"); - }); - } - - /** - * Forces the virtual thread to unmount and remount mid-task by sleeping between two child spans. - * This is the case OkHttp actually hits in practice — every blocking socket read unmounts - * the carrier, and the scope stack has to be reinstated by {@code VirtualThreadInstrumentation} - * on each remount. The fix log for this area is long (#10931, #11009, #11111), so it is worth - * pinning down with a regression test. - */ - @Test - void virtualThreadFactory_propagatesAcrossUnmount() throws Exception { - ExecutorService executor = Executors.newThreadPerTaskExecutor(Thread.ofVirtual().factory()); - try { - underParentTrace( - 2, - () -> - executor.execute( - () -> { - child(); - try { - // Sleep is a JDK 21+ carrier-unmounting parking point for virtual threads. - Thread.sleep(50); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - child(); - })); - } finally { - executor.shutdown(); - } - - writer.waitForTraces(1); - List trace = writer.get(0); - DDSpan parentSpan = findByOp(trace, "parent"); - assertNotNull(parentSpan, "parent span should exist"); - long parentSpanId = parentSpan.getSpanId(); - - long childCount = - trace.stream().filter(s -> "child".contentEquals(s.getOperationName())).count(); - assertEquals(2, childCount, "both child spans should be captured"); - - trace.stream() - .filter(s -> "child".contentEquals(s.getOperationName())) - .forEach( - s -> - assertEquals( - parentSpanId, - s.getParentId(), - "child span before and after virtual-thread unmount must both attach to parent")); - } - - private static DDSpan findByOp(List spans, String op) { - return spans.stream() - .filter(s -> op.contentEquals(s.getOperationName())) - .findFirst() - .orElse(null); - } -} diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle index e2ce1db8684..9f23fe1228b 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle @@ -23,8 +23,8 @@ tasks.named("compileVthread21TestJava", JavaCompile) { } tasks.named("vthread21Test", Test) { - javaLauncher = javaToolchains.launcherFor { - languageVersion = JavaLanguageVersion.of(21) + testJvmConstraints { + minJavaVersion = JavaVersion.VERSION_21 } } diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java index 0d9d9295b99..5629d2f9d2c 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java @@ -27,14 +27,8 @@ * caller who actually enqueued this AsyncCall. Result: under concurrent OkHttp load, {@code * okhttp.request} spans cross-contaminate between traces. * - *

The class moved between OkHttp 3.x and 4.x: - * - *

    - *
  • OkHttp 3.x — {@code okhttp3.RealCall$AsyncCall} - *
  • OkHttp 4.x — {@code okhttp3.internal.connection.RealCall$AsyncCall} - *
- * - * Both are inner classes of {@code RealCall} and both transitively implement {@link Runnable}. + *

{@code AsyncCall} is an inner class of {@code RealCall} and transitively implements {@link + * Runnable}. */ @AutoService(InstrumenterModule.class) public final class AsyncCallInstrumentation extends InstrumenterModule.Tracing @@ -51,8 +45,7 @@ public AsyncCallInstrumentation() { @Override public String[] knownMatchingTypes() { return new String[] { - "okhttp3.RealCall$AsyncCall", // OkHttp 3.x - "okhttp3.internal.connection.RealCall$AsyncCall", // OkHttp 4.x + "okhttp3.RealCall$AsyncCall", }; } From 269f42c3ead16098885734fc8989f3e22bcb98ed Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 9 Jun 2026 17:09:53 -0400 Subject: [PATCH 3/7] fix(okhttp): split AsyncCall scope capture by major version, verify 4.x/5.x Address review feedback on the virtual-thread Dispatcher trace-contamination fix: - Split the AsyncCall. scope capture into per-major instrumentations so each is tested against its own OkHttp version. okhttp-3.0 keeps the v3 type (okhttp3.RealCall$AsyncCall); a new okhttp-4.0 module matches the relocated v4+ type (okhttp3.internal.connection.RealCall$AsyncCall). The okhttp-4.0 suite runs the regression against OkHttp 4.12.0 and 5.4.0 (5.x is Kotlin Multiplatform; classes ship in okhttp-jvm, resolved via Gradle variant metadata). Muzzle covers 4.0.0 -> 5.4.0. - Both instrumentations now extend InstrumenterModule.ContextTracking (matching RunnableInstrumentation, the consumer of the state they write) rather than Tracing -- their job is scope propagation, not span creation. - Drop the redundant cachedThreadPoolDispatcher test (covered by OkHttp3AsyncTest) and document that only the concurrent test exercises the contamination path. - Cap the vthread suites at maxJavaVersion=24 to match java-concurrent-21.0, which provides the virtual-thread propagation they depend on and is only supported through JDK 24. This fixes the TimeoutExceptions on the JDK 25 / 26 ("tip") CI shards, where that propagation chain isn't active. Co-Authored-By: Claude Opus 4.8 --- .../okhttp/okhttp-3.0/build.gradle | 5 + .../okhttp3/AsyncCallInstrumentation.java | 12 +- .../OkHttpVirtualThreadDispatcherTest.java | 34 +- .../okhttp/okhttp-4.0/build.gradle | 67 ++++ .../okhttp4/AsyncCallInstrumentation.java | 65 ++++ .../OkHttpVirtualThreadDispatcherTest.java | 307 ++++++++++++++++++ settings.gradle.kts | 1 + 7 files changed, 466 insertions(+), 25 deletions(-) create mode 100644 dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle create mode 100644 dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/main/java/datadog/trace/instrumentation/okhttp4/AsyncCallInstrumentation.java create mode 100644 dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle index 9f23fe1228b..b056987e409 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle @@ -25,6 +25,11 @@ tasks.named("compileVthread21TestJava", JavaCompile) { tasks.named("vthread21Test", Test) { testJvmConstraints { minJavaVersion = JavaVersion.VERSION_21 + // Cap at 24 to match java-concurrent-21.0, which provides the virtual-thread scope propagation + // (TaskRunner/VirtualThread instrumentation) this suite relies on and is only supported through + // JDK 24. Without this, the suite runs on the JDK 25 / "tip" (26) CI shards and times out + // because that propagation chain isn't active there. + maxJavaVersion = JavaVersion.VERSION_24 } } diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java index 5629d2f9d2c..ac77c39d85f 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java @@ -28,10 +28,12 @@ * okhttp.request} spans cross-contaminate between traces. * *

{@code AsyncCall} is an inner class of {@code RealCall} and transitively implements {@link - * Runnable}. + * Runnable}. This module targets OkHttp 3.x, where it lives at {@code okhttp3.RealCall$AsyncCall}. + * OkHttp 4.x+ relocated it to {@code okhttp3.internal.connection.RealCall$AsyncCall}, which is + * handled by the separate {@code okhttp-4.0} module. */ @AutoService(InstrumenterModule.class) -public final class AsyncCallInstrumentation extends InstrumenterModule.Tracing +public final class AsyncCallInstrumentation extends InstrumenterModule.ContextTracking implements Instrumenter.ForKnownTypes, Instrumenter.HasMethodAdvice { public AsyncCallInstrumentation() { @@ -39,13 +41,17 @@ public AsyncCallInstrumentation() { // separately-toggleable feature flag (DD_TRACE_OKHTTP_ASYNC_CALL_ENABLED). The capture here // is conceptually part of the OkHttp instrumentation — if you disable OkHttp tracing, you // also disable this capture, which is the right behavior. + // + // This is a ContextTracking module (like RunnableInstrumentation, which consumes the state we + // write) rather than a Tracing module: its sole job is to propagate the captured scope through + // the shared ContextStore, not to create spans of its own. super("okhttp", "okhttp-3"); } @Override public String[] knownMatchingTypes() { return new String[] { - "okhttp3.RealCall$AsyncCall", + "okhttp3.RealCall$AsyncCall", // OkHttp 3.x }; } diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java index 6def8c6ec33..a2558906e4e 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java @@ -34,16 +34,19 @@ * {@code Executors.newCachedThreadPool(...)} to {@code * Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name(prefix, start).factory())}. * - *

The test runs both shapes against the same {@link HttpServer} mock. Inside a manually - * activated "parent" span it does {@code client.newCall(request).enqueue(callback)} and waits on a - * latch for the callback. The agent's OkHttp instrumentation injects {@code TracingInterceptor}, - * which creates the {@code okhttp.request} client span using whatever scope is active on the - * dispatcher worker. The assertions verify the client span lands under the parent — i.e., the - * dispatcher's worker thread saw the propagated scope. + *

Each test runs against the same {@link HttpServer} mock. Inside a manually activated "parent" + * span it does {@code client.newCall(request).enqueue(callback)} and waits on a latch for the + * callback. The agent's OkHttp instrumentation injects {@code TracingInterceptor}, which creates + * the {@code okhttp.request} client span using whatever scope is active on the dispatcher worker. + * The assertions verify the client span lands under the parent — i.e., the dispatcher's + * worker thread saw the propagated scope. * - *

If propagation fails for the virtual-thread shape (the failure profiling-backend is reporting) - * the {@code okhttp.request} span will either become a root span in its own trace or be parented - * under nothing, and {@code assertEquals(parentSpan.getSpanId(), okhttpSpan.getParentId())} fails. + *

{@code concurrentVirtualThreadPerTaskDispatcher_keepsEachTraceSeparate} is the regression test + * for the fix: it forces dispatcher-queue contention so calls are promoted from {@code + * Dispatcher.finished()} on a sibling's worker thread, and fails (cross-trace contamination) + * without the {@code AsyncCall.} scope capture. {@code virtualThreadPerTaskDispatcher_…} is a + * single-shot baseline: it confirms basic propagation through the virtual-thread dispatcher but + * does not by itself exercise the contamination path (single-shot requests never queue). */ class OkHttpVirtualThreadDispatcherTest extends AbstractInstrumentationTest { @@ -114,19 +117,6 @@ public void onFailure(Call call, IOException e) { } } - @Test - void cachedThreadPoolDispatcher_parentsOkHttpSpanUnderParent() throws Exception { - ExecutorService dispatcherExecutor = Executors.newCachedThreadPool(); - OkHttpClient client = buildClient(dispatcherExecutor); - try { - runUnderParent(client, new CountDownLatch(1)); - } finally { - dispatcherExecutor.shutdown(); - } - - assertOkHttpSpanParentedUnderParent(); - } - @Test void virtualThreadPerTaskDispatcher_parentsOkHttpSpanUnderParent() throws Exception { // Exact shape from profiling-backend PR#8520. diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle new file mode 100644 index 00000000000..9352fddbffd --- /dev/null +++ b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle @@ -0,0 +1,67 @@ +muzzle { + // No assertInverse: the advice references only java.lang.Runnable plus agent-bootstrap helpers, + // never an OkHttp type, so muzzle has nothing version-specific to fail on for pre-4.x. The real + // version gate is knownMatchingTypes — okhttp3.internal.connection.RealCall$AsyncCall only exists + // in 4.x+, so the instrumentation simply doesn't match on 3.x (which okhttp-3.0 handles instead). + pass { + group = "com.squareup.okhttp3" + module = "okhttp" + versions = "[4.0,)" + } +} + +apply from: "$rootDir/gradle/java.gradle" +// Use slf4j-simple for tests; logback's synchronized appenders can pin virtual-thread carriers +// when many vthreads log concurrently, which deadlocks the virtual-thread suites. +apply from: "$rootDir/gradle/slf4j-simple.gradle" + +// The only coverage here is the virtual-thread regression suite, which needs the JDK 21+ API. We +// run the same test source against both OkHttp 4.x and 5.x: +// - 4.x lives at com.squareup.okhttp3:okhttp +// - 5.x became a Kotlin-Multiplatform library; com.squareup.okhttp3:okhttp is a metadata shell and +// the JVM classes ship in com.squareup.okhttp3:okhttp-jvm (Gradle resolves this automatically +// via variant metadata). The matched type okhttp3.internal.connection.RealCall$AsyncCall is at +// the same FQN in both, so a single instrumentation covers both majors. +// Both suites share src/vthread21Test/java and are kept apart only to bind different OkHttp versions. +addTestSuiteForDir('vthread21Test4', 'vthread21Test') +addTestSuiteForDir('vthread21Test5', 'vthread21Test') + +['vthread21Test4', 'vthread21Test5'].each { suite -> + tasks.named("compile${suite.capitalize()}Java", JavaCompile) { + configureCompiler(it, 21) + } + tasks.named(suite, Test) { + testJvmConstraints { + minJavaVersion = JavaVersion.VERSION_21 + // Cap at 24 to match java-concurrent-21.0, which provides the virtual-thread scope propagation + // these suites rely on and is only supported through JDK 24. See okhttp-3.0 for details. + maxJavaVersion = JavaVersion.VERSION_24 + } + } + tasks.named("check") { + dependsOn suite + } +} + +dependencies { + compileOnly(group: 'com.squareup.okhttp3', name: 'okhttp', version: '4.0.0') + + // OkHttp 4.x (Kotlin) pulls kotlin-stdlib and okio 2.x transitively. + vthread21Test4Implementation(group: 'com.squareup.okhttp3', name: 'okhttp', version: '4.12.0') + // OkHttp 5.x resolves to okhttp-jvm via Gradle variant metadata; pulls kotlin-stdlib 2.x + okio 3.x. + vthread21Test5Implementation(group: 'com.squareup.okhttp3', name: 'okhttp', version: '5.4.0') + + // The core OkHttp tracing (OkHttp3Instrumentation -> TracingInterceptor, which creates the + // okhttp.request span) lives in the okhttp-3.0 module and applies to 4.x/5.x too. Pull it in so + // the tests exercise the full chain: core tracing + this module's AsyncCall scope capture. + vthread21Test4RuntimeOnly project(':dd-java-agent:instrumentation:okhttp:okhttp-3.0') + vthread21Test5RuntimeOnly project(':dd-java-agent:instrumentation:okhttp:okhttp-3.0') + + // Pull in the JDK-21+ concurrent / lang instrumentations so the tests install the same + // TaskRunnerInstrumentation + VirtualThreadInstrumentation chain that profiling-backend + // exercises in production. + vthread21Test4RuntimeOnly project(':dd-java-agent:instrumentation:java:java-concurrent:java-concurrent-21.0') + vthread21Test4RuntimeOnly project(':dd-java-agent:instrumentation:java:java-lang:java-lang-21.0') + vthread21Test5RuntimeOnly project(':dd-java-agent:instrumentation:java:java-concurrent:java-concurrent-21.0') + vthread21Test5RuntimeOnly project(':dd-java-agent:instrumentation:java:java-lang:java-lang-21.0') +} diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/main/java/datadog/trace/instrumentation/okhttp4/AsyncCallInstrumentation.java b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/main/java/datadog/trace/instrumentation/okhttp4/AsyncCallInstrumentation.java new file mode 100644 index 00000000000..4c71d9a35be --- /dev/null +++ b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/main/java/datadog/trace/instrumentation/okhttp4/AsyncCallInstrumentation.java @@ -0,0 +1,65 @@ +package datadog.trace.instrumentation.okhttp4; + +import static datadog.trace.bootstrap.instrumentation.java.concurrent.AdviceUtils.capture; +import static java.util.Collections.singletonMap; +import static net.bytebuddy.matcher.ElementMatchers.isConstructor; + +import com.google.auto.service.AutoService; +import datadog.trace.agent.tooling.Instrumenter; +import datadog.trace.agent.tooling.InstrumenterModule; +import datadog.trace.bootstrap.InstrumentationContext; +import datadog.trace.bootstrap.instrumentation.java.concurrent.State; +import java.util.Map; +import net.bytebuddy.asm.Advice; + +/** + * OkHttp 4.x+ variant of the {@code AsyncCall.} scope capture. Identical in behavior to the + * {@code okhttp-3.0} module's instrumentation — see that class for the full explanation of the + * dispatcher-recursion failure mode — but matches the relocated 4.x type. + * + *

{@code AsyncCall} is an inner class of {@code RealCall} and transitively implements {@link + * Runnable}. OkHttp 4.x moved {@code RealCall} from {@code okhttp3} into {@code + * okhttp3.internal.connection}, so the nested type is {@code + * okhttp3.internal.connection.RealCall$AsyncCall}. + */ +@AutoService(InstrumenterModule.class) +public final class AsyncCallInstrumentation extends InstrumenterModule.ContextTracking + implements Instrumenter.ForKnownTypes, Instrumenter.HasMethodAdvice { + + public AsyncCallInstrumentation() { + // Re-use the existing "okhttp" instrumentation name so this capture is enabled/disabled with + // OkHttp tracing as a whole, rather than introducing a separately-toggleable feature flag. + // + // This is a ContextTracking module (like RunnableInstrumentation, which consumes the state we + // write) rather than a Tracing module: its sole job is to propagate the captured scope through + // the shared ContextStore, not to create spans of its own. + super("okhttp", "okhttp-4"); + } + + @Override + public String[] knownMatchingTypes() { + return new String[] { + "okhttp3.internal.connection.RealCall$AsyncCall", // OkHttp 4.x+ + }; + } + + @Override + public Map contextStore() { + // Same Runnable -> State store that RunnableInstrumentation reads from. + return singletonMap("java.lang.Runnable", State.class.getName()); + } + + @Override + public void methodAdvice(MethodTransformer transformer) { + transformer.applyAdvice(isConstructor(), getClass().getName() + "$Construct"); + } + + public static final class Construct { + @Advice.OnMethodExit(suppress = Throwable.class) + public static void captureScope(@Advice.This Runnable asyncCall) { + // AdviceUtils.capture is a no-op when async propagation is disabled or there's no active + // span — same behavior as the rest of the concurrent instrumentation. + capture(InstrumentationContext.get(Runnable.class, State.class), asyncCall); + } + } +} diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java new file mode 100644 index 00000000000..a2558906e4e --- /dev/null +++ b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java @@ -0,0 +1,307 @@ +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.sun.net.httpserver.HttpServer; +import datadog.trace.agent.test.AbstractInstrumentationTest; +import datadog.trace.bootstrap.instrumentation.api.AgentScope; +import datadog.trace.bootstrap.instrumentation.api.AgentSpan; +import datadog.trace.bootstrap.instrumentation.api.AgentTracer; +import datadog.trace.core.DDSpan; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import okhttp3.Call; +import okhttp3.Callback; +import okhttp3.Dispatcher; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * End-to-end reproduction of profiling-backend PR#8520: swap OkHttp's Dispatcher executor from + * {@code Executors.newCachedThreadPool(...)} to {@code + * Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name(prefix, start).factory())}. + * + *

Each test runs against the same {@link HttpServer} mock. Inside a manually activated "parent" + * span it does {@code client.newCall(request).enqueue(callback)} and waits on a latch for the + * callback. The agent's OkHttp instrumentation injects {@code TracingInterceptor}, which creates + * the {@code okhttp.request} client span using whatever scope is active on the dispatcher worker. + * The assertions verify the client span lands under the parent — i.e., the dispatcher's + * worker thread saw the propagated scope. + * + *

{@code concurrentVirtualThreadPerTaskDispatcher_keepsEachTraceSeparate} is the regression test + * for the fix: it forces dispatcher-queue contention so calls are promoted from {@code + * Dispatcher.finished()} on a sibling's worker thread, and fails (cross-trace contamination) + * without the {@code AsyncCall.} scope capture. {@code virtualThreadPerTaskDispatcher_…} is a + * single-shot baseline: it confirms basic propagation through the virtual-thread dispatcher but + * does not by itself exercise the contamination path (single-shot requests never queue). + */ +class OkHttpVirtualThreadDispatcherTest extends AbstractInstrumentationTest { + + private static HttpServer mockServer; + private static String baseUrl; + + @BeforeAll + static void startServer() throws IOException { + mockServer = HttpServer.create(new InetSocketAddress("localhost", 0), 64); + mockServer.createContext( + "/ok", + exchange -> { + byte[] body = "ok".getBytes(); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); + // Default executor is single-threaded — give the server real concurrency so the test isn't + // bottlenecked on the mock backend itself. + mockServer.setExecutor(Executors.newCachedThreadPool()); + mockServer.start(); + baseUrl = + "http://" + + mockServer.getAddress().getHostString() + + ":" + + mockServer.getAddress().getPort(); + } + + @AfterAll + static void stopServer() { + if (mockServer != null) { + mockServer.stop(0); + mockServer = null; + } + } + + /** Activate a manual parent span, run the OkHttp call, wait for the okhttp.request child. */ + private void runUnderParent(OkHttpClient client, CountDownLatch done) { + AgentSpan parentSpan = AgentTracer.startSpan("test", "parent"); + try (AgentScope ignored = AgentTracer.activateSpan(parentSpan)) { + Request request = new Request.Builder().url(baseUrl + "/ok").build(); + client + .newCall(request) + .enqueue( + new Callback() { + @Override + public void onResponse(Call call, Response response) throws IOException { + response.body().close(); + done.countDown(); + } + + @Override + public void onFailure(Call call, IOException e) { + done.countDown(); + } + }); + if (!done.await(10, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting for OkHttp callback"); + } + // Wait for the okhttp.request child to finish before closing the parent scope so the trace + // collector sees a complete trace. + blockUntilChildSpansFinished(1); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } finally { + parentSpan.finish(); + } + } + + @Test + void virtualThreadPerTaskDispatcher_parentsOkHttpSpanUnderParent() throws Exception { + // Exact shape from profiling-backend PR#8520. + ExecutorService dispatcherExecutor = + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-test-", 0).factory()); + OkHttpClient client = buildClient(dispatcherExecutor); + try { + runUnderParent(client, new CountDownLatch(1)); + } finally { + dispatcherExecutor.shutdown(); + } + + assertOkHttpSpanParentedUnderParent(); + } + + /** + * Concurrent stress test: spin up N independent parent traces, each enqueueing M OkHttp requests + * through the same virtual-thread dispatcher. Dispatcher capacity is intentionally set below N*M + * so that some calls get queued and then promoted from {@code Dispatcher.finished()} running on a + * dispatcher worker thread (a different parent's worker). + * + *

If the agent captures the worker's currently-active scope when the promoted call is + * submitted — instead of the scope active where the original {@code enqueue()} happened + * — the {@code okhttp.request} span will land in the wrong parent's trace. This test fails + * loudly on that cross-contamination. + */ + @Test + void concurrentVirtualThreadPerTaskDispatcher_keepsEachTraceSeparate() throws Exception { + int parentCount = 16; + int requestsPerParent = 4; + int totalRequests = parentCount * requestsPerParent; + + ExecutorService dispatcherExecutor = + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-burst-", 0).factory()); + OkHttpClient client = buildClient(dispatcherExecutor); + // Force queue contention: capacity is much smaller than the total in-flight requests, so + // many calls will be promoted from finished() rather than direct from enqueue(). + Dispatcher dispatcher = client.dispatcher(); + dispatcher.setMaxRequests(4); + dispatcher.setMaxRequestsPerHost(4); + + ExecutorService parentRunner = Executors.newFixedThreadPool(parentCount); + CountDownLatch allParentsDone = new CountDownLatch(parentCount); + AtomicReference failure = new AtomicReference<>(); + + try { + for (int i = 0; i < parentCount; i++) { + parentRunner.submit( + () -> { + try { + runParentBurst(client, requestsPerParent); + } catch (Throwable t) { + failure.compareAndSet(null, t); + } finally { + allParentsDone.countDown(); + } + }); + } + assertTrue(allParentsDone.await(60, TimeUnit.SECONDS), "parent threads timed out"); + } finally { + parentRunner.shutdown(); + dispatcherExecutor.shutdown(); + } + + if (failure.get() != null) { + throw new AssertionError("a parent-runner thread threw", failure.get()); + } + + // Each parent produces one trace containing 1 parent span + M okhttp.request spans. + writer.waitForTraces(parentCount); + + // Map trace-id -> spans-in-that-trace and assert structure. + Map> tracesByRoot = new HashMap<>(); + for (List trace : writer) { + DDSpan parentSpan = findByOp(trace, "parent"); + assertNotNull(parentSpan, "every collected trace should have a parent span"); + tracesByRoot.put(parentSpan.getSpanId(), trace); + } + assertEquals( + parentCount, + tracesByRoot.size(), + "expected one distinct trace per parent burst (no cross-trace contamination)"); + + int totalOkhttpSpans = 0; + List contamination = new ArrayList<>(); + for (Map.Entry> entry : tracesByRoot.entrySet()) { + long parentSpanId = entry.getKey(); + List trace = entry.getValue(); + int okhttpCountInThisTrace = 0; + for (DDSpan span : trace) { + if ("okhttp.request".contentEquals(span.getOperationName())) { + okhttpCountInThisTrace++; + if (span.getParentId() != parentSpanId) { + contamination.add( + "okhttp.request span " + + span.getSpanId() + + " has parentId=" + + span.getParentId() + + " but it lives in the trace rooted at " + + parentSpanId); + } + } + } + totalOkhttpSpans += okhttpCountInThisTrace; + assertEquals( + requestsPerParent, + okhttpCountInThisTrace, + "trace rooted at parent " + parentSpanId + " has wrong child count"); + } + + assertEquals(totalRequests, totalOkhttpSpans, "total okhttp.request spans across all traces"); + assertTrue( + contamination.isEmpty(), + "found cross-parented okhttp.request spans:\n - " + String.join("\n - ", contamination)); + } + + /** + * One parent burst: M OkHttp requests under a single freshly-started parent span. Deliberately + * does not block on per-parent child-span accounting — the whole point of the test + * is to detect when children leak to a sibling's trace, and per-parent blocking would just turn + * that into a timeout instead of producing a useful assertion message. Wait for the HTTP + * callbacks (so the request actually ran), then close the parent. + */ + private void runParentBurst(OkHttpClient client, int requestsPerParent) { + AgentSpan parentSpan = AgentTracer.startSpan("test", "parent"); + try (AgentScope ignored = AgentTracer.activateSpan(parentSpan)) { + CountDownLatch done = new CountDownLatch(requestsPerParent); + for (int i = 0; i < requestsPerParent; i++) { + Request request = new Request.Builder().url(baseUrl + "/ok").build(); + client + .newCall(request) + .enqueue( + new Callback() { + @Override + public void onResponse(Call call, Response response) throws IOException { + response.body().close(); + done.countDown(); + } + + @Override + public void onFailure(Call call, IOException e) { + done.countDown(); + } + }); + } + if (!done.await(30, TimeUnit.SECONDS)) { + throw new AssertionError("timed out waiting for OkHttp callbacks"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError(e); + } finally { + parentSpan.finish(); + } + } + + private static OkHttpClient buildClient(ExecutorService dispatcherExecutor) { + Dispatcher dispatcher = new Dispatcher(dispatcherExecutor); + return new OkHttpClient.Builder().dispatcher(dispatcher).build(); + } + + private void assertOkHttpSpanParentedUnderParent() throws Exception { + writer.waitForTraces(1); + List trace = writer.get(0); + DDSpan parentSpan = findByOp(trace, "parent"); + DDSpan okhttpSpan = findByOp(trace, "okhttp.request"); + assertNotNull(parentSpan, "parent span should exist"); + assertNotNull( + okhttpSpan, + "okhttp.request client span should exist; if missing, propagation may have produced an" + + " orphan trace instead"); + assertEquals( + parentSpan.getTraceId().toLong(), + okhttpSpan.getTraceId().toLong(), + "okhttp.request span must share the parent's trace id"); + assertEquals( + parentSpan.getSpanId(), + okhttpSpan.getParentId(), + "okhttp.request span must be parented under the parent span active at enqueue() time"); + } + + private static DDSpan findByOp(List spans, String op) { + return spans.stream() + .filter(s -> op.contentEquals(s.getOperationName())) + .findFirst() + .orElse(null); + } +} diff --git a/settings.gradle.kts b/settings.gradle.kts index dda1f432e6f..74d52cca73e 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -500,6 +500,7 @@ include( ":dd-java-agent:instrumentation:ognl-appsec-3.3.2", ":dd-java-agent:instrumentation:okhttp:okhttp-2.2", ":dd-java-agent:instrumentation:okhttp:okhttp-3.0", + ":dd-java-agent:instrumentation:okhttp:okhttp-4.0", ":dd-java-agent:instrumentation:openai-java:openai-java-3.0", ":dd-java-agent:instrumentation:opensearch:opensearch-rest-1.0", ":dd-java-agent:instrumentation:opensearch:opensearch-transport-1.0", From fbcd0a1531a910462f941bed0613e975db51d295 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Tue, 23 Jun 2026 15:21:34 -0400 Subject: [PATCH 4/7] Add okhttp-4 integration to supported-configurations.json The split okhttp-4.0 module registers a new "okhttp-4" integration name, so DD_TRACE_OKHTTP_4_ENABLED must be declared in the supported configurations metadata. Co-Authored-By: Claude Opus 4.8 --- metadata/supported-configurations.json | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/metadata/supported-configurations.json b/metadata/supported-configurations.json index a42ba33a31b..fddd31b6d04 100644 --- a/metadata/supported-configurations.json +++ b/metadata/supported-configurations.json @@ -8361,6 +8361,14 @@ "aliases": ["DD_TRACE_INTEGRATION_OKHTTP_3_ENABLED", "DD_INTEGRATION_OKHTTP_3_ENABLED"] } ], + "DD_TRACE_OKHTTP_4_ENABLED": [ + { + "version": "A", + "type": "boolean", + "default": "true", + "aliases": ["DD_TRACE_INTEGRATION_OKHTTP_4_ENABLED", "DD_INTEGRATION_OKHTTP_4_ENABLED"] + } + ], "DD_TRACE_OKHTTP_ANALYTICS_ENABLED": [ { "version": "A", From a1adf8ecd8d95c6579dda0a36b82810b4df11eef Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 24 Jun 2026 09:37:01 -0400 Subject: [PATCH 5/7] Address review feedback on okhttp AsyncCall instrumentation - Scope AsyncCallInstrumentation's muzzle to a named [3.0,4.0) directive (okhttp-async-3) so the 3.x-only async-call capture no longer claims the module-wide [3.0,) range; real version gate remains knownMatchingTypes. - Remove two verbose comments amarziali flagged (the rationale lives in the vthread21Test javadoc, which documents and verifies it). - Remove the maxJavaVersion=24 cap on both okhttp vthread suites. The cap was mis-inherited from java-concurrent-21.0, whose cap exists because its own test uses StructuredTaskScope.ShutdownOnFailure (a preview API removed in JDK 25) - not an instrumentation limit. Verified okhttp 3.x/4.x/5.x vthread suites pass on JDK 25. Co-Authored-By: Claude Opus 4.8 --- .../okhttp/okhttp-3.0/build.gradle | 20 ++++++++--- .../okhttp3/AsyncCallInstrumentation.java | 34 ++++--------------- .../okhttp/okhttp-4.0/build.gradle | 3 -- 3 files changed, 21 insertions(+), 36 deletions(-) diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle index b056987e409..05cb9a6150a 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle @@ -1,10 +1,25 @@ muzzle { + // Core tracing (OkHttp3Instrumentation -> TracingInterceptor), IAST and AppSec reference OkHttp + // types and apply to 3.x/4.x/5.x. pass { + name = 'okhttp-3' group = "com.squareup.okhttp3" module = "okhttp" versions = "[3.0,)" assertInverse = true } + // AsyncCallInstrumentation is 3.x-only: the matched type okhttp3.RealCall$AsyncCall relocated to + // okhttp3.internal.connection.RealCall$AsyncCall in 4.x (handled by the okhttp-4.0 module). Its + // advice references only java.lang.Runnable, never an OkHttp type, so muzzle has nothing + // version-specific to enforce here — this directive documents the [3.0,4.0) boundary; the real + // gate is knownMatchingTypes. No assertInverse for the same reason: with no OkHttp reference + // muzzle would pass outside the range (nothing to be missing), contradicting the assertion. + pass { + name = 'okhttp-async-3' + group = "com.squareup.okhttp3" + module = "okhttp" + versions = "[3.0,4.0)" + } } apply from: "$rootDir/gradle/java.gradle" @@ -25,11 +40,6 @@ tasks.named("compileVthread21TestJava", JavaCompile) { tasks.named("vthread21Test", Test) { testJvmConstraints { minJavaVersion = JavaVersion.VERSION_21 - // Cap at 24 to match java-concurrent-21.0, which provides the virtual-thread scope propagation - // (TaskRunner/VirtualThread instrumentation) this suite relies on and is only supported through - // JDK 24. Without this, the suite runs on the JDK 25 / "tip" (26) CI shards and times out - // because that propagation chain isn't active there. - maxJavaVersion = JavaVersion.VERSION_24 } } diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java index ac77c39d85f..976a92bb9f0 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/main/java/datadog/trace/instrumentation/okhttp3/AsyncCallInstrumentation.java @@ -12,42 +12,20 @@ import java.util.Map; import net.bytebuddy.asm.Advice; -/** - * Captures the active scope at {@code AsyncCall.} (i.e. the moment {@code - * Call.enqueue(callback)} was invoked on the user's thread) and stores it in the {@code - * ContextStore} shared with the rest of the concurrent instrumentation. {@code - * RunnableInstrumentation} then re-activates that scope on {@code AsyncCall.run()} entry, which - * overrides whatever scope {@code TaskRunner.run()} (or {@code beforeExecute}) put in place from - * the dispatcher's worker thread. - * - *

Without this, {@code TaskRunnerInstrumentation} captures whatever scope happens to be active - * on the worker thread when {@code Dispatcher.promoteAndExecute()} dequeues and submits the call — - * and when promotion runs from inside {@code Dispatcher.finished()} (i.e. recursively from a - * different AsyncCall's run()), that scope belongs to the finishing call, not to the - * caller who actually enqueued this AsyncCall. Result: under concurrent OkHttp load, {@code - * okhttp.request} spans cross-contaminate between traces. - * - *

{@code AsyncCall} is an inner class of {@code RealCall} and transitively implements {@link - * Runnable}. This module targets OkHttp 3.x, where it lives at {@code okhttp3.RealCall$AsyncCall}. - * OkHttp 4.x+ relocated it to {@code okhttp3.internal.connection.RealCall$AsyncCall}, which is - * handled by the separate {@code okhttp-4.0} module. - */ @AutoService(InstrumenterModule.class) public final class AsyncCallInstrumentation extends InstrumenterModule.ContextTracking implements Instrumenter.ForKnownTypes, Instrumenter.HasMethodAdvice { public AsyncCallInstrumentation() { - // Re-use the existing "okhttp" / "okhttp-3" instrumentation names so we don't introduce a - // separately-toggleable feature flag (DD_TRACE_OKHTTP_ASYNC_CALL_ENABLED). The capture here - // is conceptually part of the OkHttp instrumentation — if you disable OkHttp tracing, you - // also disable this capture, which is the right behavior. - // - // This is a ContextTracking module (like RunnableInstrumentation, which consumes the state we - // write) rather than a Tracing module: its sole job is to propagate the captured scope through - // the shared ContextStore, not to create spans of its own. super("okhttp", "okhttp-3"); } + @Override + public String muzzleDirective() { + // 3.x only: okhttp3.RealCall$AsyncCall relocated in 4.x (handled by the okhttp-4.0 module). + return "okhttp-async-3"; + } + @Override public String[] knownMatchingTypes() { return new String[] { diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle index 9352fddbffd..9564342842a 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle @@ -33,9 +33,6 @@ addTestSuiteForDir('vthread21Test5', 'vthread21Test') tasks.named(suite, Test) { testJvmConstraints { minJavaVersion = JavaVersion.VERSION_21 - // Cap at 24 to match java-concurrent-21.0, which provides the virtual-thread scope propagation - // these suites rely on and is only supported through JDK 24. See okhttp-3.0 for details. - maxJavaVersion = JavaVersion.VERSION_24 } } tasks.named("check") { From d17f52367c275c336ebe98388d1d639742030d01 Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 24 Jun 2026 11:01:49 -0400 Subject: [PATCH 6/7] Restore JDK 24 cap on okhttp vthread suites (CI stability) CI showed the suites fail on the JDK 25 / tip test_inst shards once the cap was lifted. The virtual-thread instrumentation does work on 25 (the 3.x/4.x/5.x vthread suites pass locally, including the contention regression), so this is a CI/test-stability issue (the contention test's timing under shard load), not an instrumentation limit. Restoring maxJavaVersion=24 with a corrected comment; lifting it needs the suite made stable on 25+ first (separate follow-up). Co-Authored-By: Claude Opus 4.8 --- .../instrumentation/okhttp/okhttp-3.0/build.gradle | 6 ++++++ .../instrumentation/okhttp/okhttp-4.0/build.gradle | 4 ++++ 2 files changed, 10 insertions(+) diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle index 05cb9a6150a..a533aa70ab8 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/build.gradle @@ -40,6 +40,12 @@ tasks.named("compileVthread21TestJava", JavaCompile) { tasks.named("vthread21Test", Test) { testJvmConstraints { minJavaVersion = JavaVersion.VERSION_21 + // Cap at 24: the suite fails on the JDK 25 / "tip" (26) CI shards. The virtual-thread + // instrumentation itself works on 25 (the okhttp 3.x/4.x/5.x vthread suites pass locally, + // including the contention regression), so this is a CI/test-stability issue (the contention + // test's timing under shard load), not an instrumentation limit. Lift once the suite is made + // stable on 25+. + maxJavaVersion = JavaVersion.VERSION_24 } } diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle index 9564342842a..6cda6545a88 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle +++ b/dd-java-agent/instrumentation/okhttp/okhttp-4.0/build.gradle @@ -33,6 +33,10 @@ addTestSuiteForDir('vthread21Test5', 'vthread21Test') tasks.named(suite, Test) { testJvmConstraints { minJavaVersion = JavaVersion.VERSION_21 + // Cap at 24: suite fails on the JDK 25/tip CI shards (CI/test-stability, not an + // instrumentation limit — the vthread instrumentation works on 25; see okhttp-3.0). Lift + // once stable on 25+. + maxJavaVersion = JavaVersion.VERSION_24 } } tasks.named("check") { From 664f90c9e3de2b4e4bab864b5eb283b75ffe141b Mon Sep 17 00:00:00 2001 From: Douglas Q Hawkins Date: Wed, 24 Jun 2026 11:29:57 -0400 Subject: [PATCH 7/7] Add deterministic promotion-from-finished() contamination test MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The existing concurrent test relies on stochastic queue contention to trigger promotion from Dispatcher.finished() on a sibling's worker — which makes it flaky under CI-shard load (the reason the suite is capped at JDK 24). This adds a deterministic counterpart: dispatcher capacity 1, call_B (parent B) occupies the slot and blocks at a gated endpoint, call_A (parent A) is enqueued and necessarily queues, then releasing the gate promotes call_A from inside call_B's finished() with scope B active. Asserts call_A's okhttp.request span parents under A. No bursts, no timing windows — fails identically every run. Verified: passes with the AsyncCall. capture; fails without it (advice disabled) with the cross-trace contamination message. The stress test stays as an integration sanity check. Co-Authored-By: Claude Opus 4.8 --- .../OkHttpVirtualThreadDispatcherTest.java | 137 ++++++++++++++++++ 1 file changed, 137 insertions(+) diff --git a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java index a2558906e4e..2838199029e 100644 --- a/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java +++ b/dd-java-agent/instrumentation/okhttp/okhttp-3.0/src/vthread21Test/java/OkHttpVirtualThreadDispatcherTest.java @@ -47,12 +47,24 @@ * without the {@code AsyncCall.} scope capture. {@code virtualThreadPerTaskDispatcher_…} is a * single-shot baseline: it confirms basic propagation through the virtual-thread dispatcher but * does not by itself exercise the contamination path (single-shot requests never queue). + * + *

{@code promotedFromFinished_keepsEnqueuingTraceNotFinishingTrace} is the deterministic + * regression guard: it constructs the same promotion-from-{@code finished()} contamination with two + * calls and a capacity-1 gate, so it fails identically every run instead of relying on stochastic + * contention. The concurrent stress test above stays as an integration sanity check. */ class OkHttpVirtualThreadDispatcherTest extends AbstractInstrumentationTest { private static HttpServer mockServer; private static String baseUrl; + // Gate for the deterministic promotion test: the /gated handler signals when it begins serving + // and + // then blocks until released, letting a test hold one call "running" (occupying a capacity-1 + // dispatcher slot) while a second call is enqueued and queues behind it. + private static volatile CountDownLatch gateServing; + private static volatile CountDownLatch gateRelease; + @BeforeAll static void startServer() throws IOException { mockServer = HttpServer.create(new InetSocketAddress("localhost", 0), 64); @@ -64,6 +76,27 @@ static void startServer() throws IOException { exchange.getResponseBody().write(body); exchange.close(); }); + // Gated endpoint: signals when it starts serving, then blocks until the test releases it. + mockServer.createContext( + "/gated", + exchange -> { + CountDownLatch serving = gateServing; + if (serving != null) { + serving.countDown(); + } + try { + CountDownLatch release = gateRelease; + if (release != null) { + release.await(30, TimeUnit.SECONDS); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + byte[] body = "ok".getBytes(); + exchange.sendResponseHeaders(200, body.length); + exchange.getResponseBody().write(body); + exchange.close(); + }); // Default executor is single-threaded — give the server real concurrency so the test isn't // bottlenecked on the mock backend itself. mockServer.setExecutor(Executors.newCachedThreadPool()); @@ -233,6 +266,95 @@ void concurrentVirtualThreadPerTaskDispatcher_keepsEachTraceSeparate() throws Ex "found cross-parented okhttp.request spans:\n - " + String.join("\n - ", contamination)); } + /** + * Deterministic counterpart to the stress test: rather than relying on stochastic queue + * contention, it constructs the promotion-from-{@code finished()} path explicitly. With + * dispatcher capacity 1, {@code call_B} (parent B) occupies the only slot and blocks at the gated + * endpoint; {@code call_A} (parent A) is then enqueued and necessarily queues. Releasing the gate + * lets {@code call_B} finish, and its {@code Dispatcher.finished()} promotes {@code call_A} from + * inside {@code call_B}'s worker — with scope B active there. Without the {@code + * AsyncCall.} capture {@code call_A}'s {@code okhttp.request} span lands under parent B; + * with it, under parent A. No bursts, no timing windows — it fails the same way every run, so it + * is the reliable regression guard (the stress test above remains as an integration sanity + * check). + */ + @Test + void promotedFromFinished_keepsEnqueuingTraceNotFinishingTrace() throws Exception { + ExecutorService dispatcherExecutor = + Executors.newThreadPerTaskExecutor(Thread.ofVirtual().name("okhttp-promote-", 0).factory()); + OkHttpClient client = buildClient(dispatcherExecutor); + Dispatcher dispatcher = client.dispatcher(); + dispatcher.setMaxRequests(1); + dispatcher.setMaxRequestsPerHost(1); + + gateServing = new CountDownLatch(1); + gateRelease = new CountDownLatch(1); + CountDownLatch bDone = new CountDownLatch(1); + CountDownLatch aDone = new CountDownLatch(1); + + AgentSpan parentB = AgentTracer.startSpan("test", "parentB"); + AgentSpan parentA = AgentTracer.startSpan("test", "parentA"); + try { + // call_B (the "finishing" trace) occupies the single slot and blocks at /gated. + try (AgentScope ignored = AgentTracer.activateSpan(parentB)) { + client + .newCall(new Request.Builder().url(baseUrl + "/gated").build()) + .enqueue(countdownCallback(bDone)); + } + assertTrue(gateServing.await(10, TimeUnit.SECONDS), "call_B never reached the server"); + + // call_A (the "enqueuing" trace) must queue: the only slot is held by call_B. + try (AgentScope ignored = AgentTracer.activateSpan(parentA)) { + client + .newCall(new Request.Builder().url(baseUrl + "/ok").build()) + .enqueue(countdownCallback(aDone)); + } + + // Release call_B -> finished() -> promoteAndExecute() runs call_A from B's worker (scope B + // active there). The fix must keep call_A parented under A regardless. + gateRelease.countDown(); + + assertTrue(bDone.await(10, TimeUnit.SECONDS), "call_B callback timed out"); + assertTrue(aDone.await(10, TimeUnit.SECONDS), "call_A callback timed out"); + } finally { + parentA.finish(); + parentB.finish(); + dispatcherExecutor.shutdown(); + } + + // One trace per parent; each must hold its own okhttp.request span. + writer.waitForTraces(2); + DDSpan parentASpan = null; + DDSpan parentBSpan = null; + DDSpan aOkhttp = null; + DDSpan bOkhttp = null; + for (List trace : writer) { + DDSpan ok = findByOp(trace, "okhttp.request"); + if (findByOp(trace, "parentA") != null) { + parentASpan = findByOp(trace, "parentA"); + aOkhttp = ok; + } else if (findByOp(trace, "parentB") != null) { + parentBSpan = findByOp(trace, "parentB"); + bOkhttp = ok; + } + } + + assertNotNull(parentASpan, "parentA trace should exist"); + assertNotNull(parentBSpan, "parentB trace should exist"); + assertNotNull( + aOkhttp, + "call_A's okhttp.request must live in parentA's trace; if null it leaked to parentB's trace" + + " (promotion captured the finishing call's scope instead of the enqueuing scope)"); + assertNotNull(bOkhttp, "call_B's okhttp.request should live in parentB's trace"); + assertEquals( + parentASpan.getSpanId(), + aOkhttp.getParentId(), + "call_A was promoted from call_B's finished() under scope B, but its span must parent under A" + + " (the scope active where call_A was enqueued)"); + assertEquals( + parentBSpan.getSpanId(), bOkhttp.getParentId(), "call_B's span must parent under B"); + } + /** * One parent burst: M OkHttp requests under a single freshly-started parent span. Deliberately * does not block on per-parent child-span accounting — the whole point of the test @@ -278,6 +400,21 @@ private static OkHttpClient buildClient(ExecutorService dispatcherExecutor) { return new OkHttpClient.Builder().dispatcher(dispatcher).build(); } + private static Callback countdownCallback(CountDownLatch done) { + return new Callback() { + @Override + public void onResponse(Call call, Response response) throws IOException { + response.body().close(); + done.countDown(); + } + + @Override + public void onFailure(Call call, IOException e) { + done.countDown(); + } + }; + } + private void assertOkHttpSpanParentedUnderParent() throws Exception { writer.waitForTraces(1); List trace = writer.get(0);