Remove executor queue offer check
MODIFY
auto-api/src/main/java/io/opentelemetry/instrumentation/auto/api/concurrent/ExecutorInstrumentationUtils.java
auto-api/src/main/java/io/opentelemetry/instrumentation/auto/api/concurrent/ExecutorInstrumentationUtils.java
@@ -21,7 +21,6 @@ import static io.opentelemetry.instrumentation.auto.api.concurrent.AdviceUtils.T
import io.grpc.Context;
import io.opentelemetry.instrumentation.api.config.Config;
import io.opentelemetry.instrumentation.auto.api.ContextStore;
-import io.opentelemetry.instrumentation.auto.api.WeakMap;
import io.opentelemetry.trace.Span;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
@@ -34,9 +33,6 @@ public class ExecutorInstrumentationUtils {
private static final Logger log = LoggerFactory.getLogger(ExecutorInstrumentationUtils.class);
- private static final WeakMap<Executor, Boolean> EXECUTORS_DISABLED_FOR_WRAPPED_TASKS =
- WeakMap.Provider.newWeakMap();
-
/**
* Checks if given task should get state attached.
*
@@ -54,7 +50,6 @@ public class ExecutorInstrumentationUtils {
Class<?> enclosingClass = taskClass.getEnclosingClass();
return span.getContext().isValid()
- && !ExecutorInstrumentationUtils.isExecutorDisabledForThisTask(executor, task)
// TODO Workaround for
// https://github.com/open-telemetry/opentelemetry-java-instrumentation/issues/787
&& !taskClass.getName().equals("org.apache.tomcat.util.net.NioEndpoint$SocketProcessor")
@@ -109,20 +104,4 @@ public class ExecutorInstrumentationUtils {
state.clearParentContext();
}
}
-
- public static void disableExecutorForWrappedTasks(Executor executor) {
- log.debug("Disabling Executor tracing for wrapped tasks for instance {}", executor);
- EXECUTORS_DISABLED_FOR_WRAPPED_TASKS.put(executor, true);
- }
-
- /**
- * Check if Executor can accept given task.
- *
- * <p>Disabled executors cannot accept wrapped tasks, non wrapped tasks (i.e. tasks with injected
- * fields) should still work fine.
- */
- public static boolean isExecutorDisabledForThisTask(Executor executor, Object task) {
- return (task instanceof RunnableWrapper || task instanceof CallableWrapper)
- && EXECUTORS_DISABLED_FOR_WRAPPED_TASKS.containsKey(executor);
- }
}

MODIFY
instrumentation/java-concurrent/java-concurrent.gradle
instrumentation/java-concurrent/java-concurrent.gradle
+ext {
+ // TODO (trask) currently this is only needed for LambdaGen
+ // revisit if we don't end up dropping Java 7
+ minJavaVersionForTests = JavaVersion.VERSION_1_8
+}
+
apply from: "$rootDir/gradle/instrumentation.gradle"
apply from: "$rootDir/gradle/test-with-scala.gradle"
apply plugin: 'org.unbroken-dome.test-sets'

MODIFY
instrumentation/java-concurrent/src/main/java/io/opentelemetry/instrumentation/auto/javaconcurrent/JavaExecutorInstrumentation.java
instrumentation/java-concurrent/src/main/java/io/opentelemetry/instrumentation/auto/javaconcurrent/JavaExecutorInstrumentation.java
@@ -217,16 +217,12 @@ public final class JavaExecutorInstrumentation extends AbstractExecutorInstrumen
for (Callable<?> task : tasks) {
if (task != null) {
Callable newTask = CallableWrapper.wrapIfNeeded(task);
- if (ExecutorInstrumentationUtils.isExecutorDisabledForThisTask(executor, newTask)) {
- wrappedTasks.add(task);
- } else {
wrappedTasks.add(newTask);
ContextStore<Callable, State> contextStore =
InstrumentationContext.get(Callable.class, State.class);
ExecutorInstrumentationUtils.setupState(contextStore, newTask, Context.current());
}
}
- }
tasks = wrappedTasks;
return tasks;
}

DELETE
instrumentation/java-concurrent/src/main/java/io/opentelemetry/instrumentation/auto/javaconcurrent/ThreadPoolExecutorInstrumentation.java
-/*
- * Copyright The OpenTelemetry Authors
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package io.opentelemetry.instrumentation.auto.javaconcurrent;
-
-import static java.util.Collections.singletonMap;
-import static net.bytebuddy.matcher.ElementMatchers.isConstructor;
-import static net.bytebuddy.matcher.ElementMatchers.named;
-import static net.bytebuddy.matcher.ElementMatchers.takesArgument;
-import static net.bytebuddy.matcher.ElementMatchers.takesArguments;
-
-import com.google.auto.service.AutoService;
-import io.opentelemetry.instrumentation.auto.api.concurrent.ExecutorInstrumentationUtils;
-import io.opentelemetry.instrumentation.auto.api.concurrent.GenericRunnable;
-import io.opentelemetry.javaagent.tooling.Instrumenter;
-import java.util.Map;
-import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.ThreadPoolExecutor;
-import net.bytebuddy.asm.Advice;
-import net.bytebuddy.description.method.MethodDescription;
-import net.bytebuddy.description.type.TypeDescription;
-import net.bytebuddy.matcher.ElementMatcher;
-
-@AutoService(Instrumenter.class)
-/**
- * Disable instrumentation for executors that cannot take our runnable wrappers.
- *
- * <p>FIXME: We should remove this once https://github.com/raphw/byte-buddy/issues/558 is fixed
- */
-public class ThreadPoolExecutorInstrumentation extends Instrumenter.Default {
-
- public ThreadPoolExecutorInstrumentation() {
- super(AbstractExecutorInstrumentation.EXEC_NAME);
- }
-
- @Override
- public ElementMatcher<? super TypeDescription> typeMatcher() {
- return named("java.util.concurrent.ThreadPoolExecutor");
- }
-
- @Override
- public Map<? extends ElementMatcher<? super MethodDescription>, String> transformers() {
- return singletonMap(
- isConstructor()
- .and(takesArgument(4, named("java.util.concurrent.BlockingQueue")))
- .and(takesArguments(7)),
- ThreadPoolExecutorInstrumentation.class.getName() + "$ThreadPoolExecutorAdvice");
- }
-
- public static class ThreadPoolExecutorAdvice {
- @Advice.OnMethodExit(suppress = Throwable.class)
- public static void disableIfQueueWrongType(
- @Advice.This ThreadPoolExecutor executor,
- @Advice.Argument(4) BlockingQueue<Runnable> queue) {
-
- if (queue.isEmpty()) {
- try {
- queue.offer(new GenericRunnable());
- queue.clear(); // Remove the Runnable we just added.
- } catch (ClassCastException | IllegalArgumentException e) {
- // These errors indicate the queue is fundamentally incompatible with wrapped runnables.
- // We must disable the executor instance to avoid passing wrapped runnables later.
- ExecutorInstrumentationUtils.disableExecutorForWrappedTasks(executor);
- } catch (Exception e) {
- // Other errors might indicate the queue is not fully initialized.
- // We might want to disable for those too, but for now just ignore.
- }
- }
- }
- }
-}

MODIFY
instrumentation/java-concurrent/src/test/groovy/ExecutorInstrumentationTest.groovy
instrumentation/java-concurrent/src/test/groovy/ExecutorInstrumentationTest.groovy
@@ -19,8 +19,6 @@ import static org.junit.Assume.assumeTrue
import io.opentelemetry.auto.test.AgentTestRunner
import io.opentelemetry.auto.test.utils.ConfigUtils
-import io.opentelemetry.instrumentation.auto.api.concurrent.CallableWrapper
-import io.opentelemetry.instrumentation.auto.api.concurrent.RunnableWrapper
import io.opentelemetry.sdk.trace.data.SpanData
import java.lang.reflect.InvocationTargetException
import java.util.concurrent.AbstractExecutorService
@@ -157,7 +155,7 @@ class ExecutorInstrumentationTest extends AgentTestRunner {
"execute Runnable" | executeRunnable | java7SafeCompletableFutureThreadPerTaskExecutor()
}
- def "#poolImpl '#name' disabled wrapping"() {
+ def "#poolImpl '#name' wrap lambdas"() {
setup:
def pool = poolImpl
def m = method
@@ -174,30 +172,28 @@ class ExecutorInstrumentationTest extends AgentTestRunner {
}.run()
// We block in child to make sure spans close in predictable order
child.unblock()
+ child.waitForCompletion()
- TEST_WRITER.waitForTraces(2)
+ TEST_WRITER.waitForTraces(1)
+ List<SpanData> trace = TEST_WRITER.traces[0]
expect:
- TEST_WRITER.traces.size() == 2
- TEST_WRITER.traces[0].size() == 1
- TEST_WRITER.traces[0][0].name == "parent"
- TEST_WRITER.traces[1].size() == 1
- TEST_WRITER.traces[1][0].name == "asyncChild"
+ TEST_WRITER.traces.size() == 1
+ trace.size() == 2
+ trace.get(0).name == "parent"
+ trace.get(1).name == "asyncChild"
+ trace.get(1).parentSpanId == trace.get(0).spanId
cleanup:
pool?.shutdown()
where:
- // Scheduled executor cannot accept wrapped tasks
- // TODO: we should have a test that passes lambda, but this is hard
- // because this requires tests to be run in java8+ only.
- // Instead we 'hand-wrap' tasks in this test.
name | method | wrap | poolImpl
- "execute Runnable" | executeRunnable | { new RunnableWrapper(it) } | new ScheduledThreadPoolExecutor(1)
- "submit Runnable" | submitRunnable | { new RunnableWrapper(it) } | new ScheduledThreadPoolExecutor(1)
- "submit Callable" | submitCallable | { new CallableWrapper(it) } | new ScheduledThreadPoolExecutor(1)
- "schedule Runnable" | scheduleRunnable | { new RunnableWrapper(it) } | new ScheduledThreadPoolExecutor(1)
- "schedule Callable" | scheduleCallable | { new CallableWrapper(it) } | new ScheduledThreadPoolExecutor(1)
+ "execute Runnable" | executeRunnable | { LambdaGen.wrapRunnable(it) } | new ScheduledThreadPoolExecutor(1)
+ "submit Runnable" | submitRunnable | { LambdaGen.wrapRunnable(it) } | new ScheduledThreadPoolExecutor(1)
+ "submit Callable" | submitCallable | { LambdaGen.wrapCallable(it) } | new ScheduledThreadPoolExecutor(1)
+ "schedule Runnable" | scheduleRunnable | { LambdaGen.wrapRunnable(it) } | new ScheduledThreadPoolExecutor(1)
+ "schedule Callable" | scheduleCallable | { LambdaGen.wrapCallable(it) } | new ScheduledThreadPoolExecutor(1)
}
def "#poolImpl '#name' reports after canceled jobs"() {

RENAME
auto-api/src/main/java/io/opentelemetry/instrumentation/auto/api/concurrent/GenericRunnable.java
instrumentation/java-concurrent/src/test/java/LambdaGen.java
* limitations under the License.
*/
-package io.opentelemetry.instrumentation.auto.api.concurrent;
+import java.util.concurrent.Callable;
-/** Used by ThreadPoolExecutorInstrumentation to check executor support */
-public class GenericRunnable implements Runnable {
+class LambdaGen {
- @Override
- public void run() {}
+ static Callable<?> wrapCallable(Callable<?> callable) {
+ return () -> callable.call();
+ }
+
+ static Runnable wrapRunnable(Runnable runnable) {
+ return () -> runnable.run();
+ }
}

Remove executor queue offer check (#1090)