Add thread propagation debugger
MODIFY
agent-bootstrap/src/main/java/io/opentelemetry/auto/bootstrap/instrumentation/java/concurrent/ExecutorInstrumentationUtils.java
agent-bootstrap/src/main/java/io/opentelemetry/auto/bootstrap/instrumentation/java/concurrent/ExecutorInstrumentationUtils.java
@@ -22,6 +22,8 @@ import io.grpc.Context;
import io.opentelemetry.auto.bootstrap.ContextStore;
import io.opentelemetry.auto.bootstrap.WeakMap;
import io.opentelemetry.trace.Span;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,6 +31,13 @@ import org.slf4j.LoggerFactory;
/** Utils for concurrent instrumentations. */
public class ExecutorInstrumentationUtils {
+ // locations where the context was propagated to another thread (tracking multiple steps is
+ // helpful in akka where there is so much recursive async spawning of new work)
+ public static final Context.Key<List<StackTraceElement[]>> THREAD_PROPAGATION_LOCATIONS =
+ Context.key("thread-propagation-locations");
+ private static final boolean THREAD_PROPAGATION_DEBUGGER =
+ Boolean.getBoolean("otel.threadPropagationDebugger");
+
private static final Logger log = LoggerFactory.getLogger(ExecutorInstrumentationUtils.class);
private static final WeakMap<Executor, Boolean> EXECUTORS_DISABLED_FOR_WRAPPED_TASKS =
@@ -70,12 +79,20 @@ public class ExecutorInstrumentationUtils {
* @param <T> task class type
* @param contextStore context storage
* @param task task instance
- * @param context current span
+ * @param context current context
* @return new state
*/
public static <T> State setupState(
- final ContextStore<T, State> contextStore, final T task, final Context context) {
+ final ContextStore<T, State> contextStore, final T task, Context context) {
State state = contextStore.putIfAbsent(task, State.FACTORY);
+ if (THREAD_PROPAGATION_DEBUGGER) {
+ List<StackTraceElement[]> location = THREAD_PROPAGATION_LOCATIONS.get(context);
+ if (location == null) {
+ location = new CopyOnWriteArrayList<>();
+ context = context.withValue(THREAD_PROPAGATION_LOCATIONS, location);
+ }
+ location.add(0, new Exception().getStackTrace());
+ }
state.setParentSpan(context);
return state;
}

MODIFY
testing-common/src/main/groovy/io/opentelemetry/auto/test/AgentTestRunner.java
testing-common/src/main/groovy/io/opentelemetry/auto/test/AgentTestRunner.java
@@ -77,6 +77,11 @@ public abstract class AgentTestRunner extends AgentSpecification {
private static final org.slf4j.Logger log = LoggerFactory.getLogger(AgentTestRunner.class);
+ static {
+ // always run with the thread propagation debugger to help track down sporadic test failures
+ System.setProperty("otel.threadPropagationDebugger", "true");
+ }
+
/**
* For test runs, agent's global tracer will report to this list writer.
*

MODIFY
testing-common/src/main/groovy/io/opentelemetry/auto/test/base/HttpServerTestAdvice.java
testing-common/src/main/groovy/io/opentelemetry/auto/test/base/HttpServerTestAdvice.java
@@ -18,17 +18,27 @@ package io.opentelemetry.auto.test.base;
import static io.opentelemetry.trace.TracingContextUtils.currentContextWith;
+import io.grpc.Context;
import io.opentelemetry.OpenTelemetry;
+import io.opentelemetry.auto.bootstrap.instrumentation.java.concurrent.ExecutorInstrumentationUtils;
import io.opentelemetry.auto.instrumentation.api.SpanWithScope;
import io.opentelemetry.trace.Span;
import io.opentelemetry.trace.Tracer;
+import java.util.Iterator;
+import java.util.List;
import net.bytebuddy.asm.Advice;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public abstract class HttpServerTestAdvice {
+ // needs to be public otherwise IllegalAccessError from inlined advice below
public static final Tracer TRACER =
OpenTelemetry.getTracerProvider().get("io.opentelemetry.auto");
+ // needs to be public otherwise IllegalAccessError from inlined advice below
+ public static final Logger log = LoggerFactory.getLogger(HttpServerTestAdvice.class);
+
/**
* This advice should be applied at the root of a http server request to validate the
* instrumentation correctly ignores other traces.
@@ -40,22 +50,25 @@ public abstract class HttpServerTestAdvice {
// Skip if not running the HttpServerTest.
return null;
}
- Span span = TRACER.spanBuilder("TEST_SPAN").startSpan();
- /*
- NB! This is a "hack" to debug flaky tests.
- If we have a valid parent span at this point, then tests will fail, because they expect
- "TEST_SPAN" from above to be a root span.
- This failure is good because we must not have a valid span at this point.
- Thus failure signal a span leak and we should investigate that.
- But we want to have some debug information about this currently active span.
- As span is not readable by default, the only way to obtain that information is pass
- it to InMemorySpanExporter used by tests.
- */
- Span parentSpan = TRACER.getCurrentSpan();
- if (parentSpan.getContext().isValid()) {
- parentSpan.end();
+ List<StackTraceElement[]> location =
+ ExecutorInstrumentationUtils.THREAD_PROPAGATION_LOCATIONS.get(Context.current());
+ if (location != null) {
+ StringBuilder sb = new StringBuilder();
+ Iterator<StackTraceElement[]> i = location.iterator();
+ while (i.hasNext()) {
+ for (StackTraceElement ste : i.next()) {
+ sb.append("\n");
+ sb.append(ste);
+ }
+ if (i.hasNext()) {
+ sb.append("\nwhich was propagated from:");
}
+ }
+ log.error("a context leak was detected. it was propagated from:{}", sb);
+ }
+
+ Span span = TRACER.spanBuilder("TEST_SPAN").startSpan();
return new SpanWithScope(span, currentContextWith(span));
}

Add thread propagation debugger (#860)