Instrument Runnables/Callables instead of wrapping them

CHANGELOG.asciidoc

@@ -28,6 +28,9 @@ endif::[]
 
 [float]
 ===== Features
+* Overhaul of the `ExecutorService` instrumentation that avoids issues like ``ClassCastException``s - {pull}1206[#1206]
+* Support for `ForJoinPool` and `ScheduledExecutorService` (see <<supported-async-frameworks>>)
+* Support for `ExecutorService#invokeAny` and `ExecutorService#invokeAll`
 
 [float]
 ===== Bug fixes

apm-agent-core/src/main/java/co/elastic/apm/agent/bci/ElasticApmAgent.java

@@ -34,10 +34,10 @@
 import co.elastic.apm.agent.bci.bytebuddy.SoftlyReferencingTypePoolCache;
 import co.elastic.apm.agent.bci.methodmatching.MethodMatcher;
 import co.elastic.apm.agent.bci.methodmatching.TraceMethodInstrumentation;
+import co.elastic.apm.agent.collections.WeakMapSupplier;
 import co.elastic.apm.agent.configuration.CoreConfiguration;
 import co.elastic.apm.agent.impl.ElasticApmTracer;
 import co.elastic.apm.agent.impl.ElasticApmTracerBuilder;
-import co.elastic.apm.agent.matcher.WildcardMatcher;
 import co.elastic.apm.agent.util.DependencyInjectingServiceLoader;
 import co.elastic.apm.agent.util.ExecutorUtils;
 import co.elastic.apm.agent.util.ThreadUtils;
@@ -82,9 +82,11 @@
 import static co.elastic.apm.agent.bci.ElasticApmInstrumentation.tracer;
 import static co.elastic.apm.agent.bci.bytebuddy.ClassLoaderNameMatcher.classLoaderWithName;
 import static co.elastic.apm.agent.bci.bytebuddy.ClassLoaderNameMatcher.isReflectionClassLoader;
+import static co.elastic.apm.agent.bci.bytebuddy.CustomElementMatchers.anyMatch;
 import static net.bytebuddy.asm.Advice.ExceptionHandler.Default.PRINTING;
 import static net.bytebuddy.matcher.ElementMatchers.any;
 import static net.bytebuddy.matcher.ElementMatchers.is;
+import static net.bytebuddy.matcher.ElementMatchers.isAbstract;
 import static net.bytebuddy.matcher.ElementMatchers.isInterface;
 import static net.bytebuddy.matcher.ElementMatchers.nameContains;
 import static net.bytebuddy.matcher.ElementMatchers.nameEndsWith;
@@ -104,7 +106,7 @@ public class ElasticApmAgent {
     @Nullable
     private static ResettableClassFileTransformer resettableClassFileTransformer;
     private static final List<ResettableClassFileTransformer> dynamicClassFileTransformers = new ArrayList<>();
-    private static final WeakConcurrentMap<Class<?>, Set<Collection<Class<? extends ElasticApmInstrumentation>>>> dynamicallyInstrumentedClasses = new WeakConcurrentMap.WithInlinedExpunction<>();
+    private static final WeakConcurrentMap<Class<?>, Set<Collection<Class<? extends ElasticApmInstrumentation>>>> dynamicallyInstrumentedClasses = WeakMapSupplier.createMap();
     @Nullable
     private static File agentJarFile;
 
@@ -249,7 +251,7 @@ private static AgentBuilder applyAdvice(final ElasticApmTracer tracer, final Age
         final ElementMatcher.Junction<ClassLoader> classLoaderMatcher = instrumentation.getClassLoaderMatcher();
         final ElementMatcher<? super NamedElement> typeMatcherPreFilter = instrumentation.getTypeMatcherPreFilter();
         final ElementMatcher.Junction<ProtectionDomain> versionPostFilter = instrumentation.getImplementationVersionPostFilter();
-        final ElementMatcher<? super MethodDescription> methodMatcher = instrumentation.getMethodMatcher();
+        final ElementMatcher<? super MethodDescription> methodMatcher = new ElementMatcher.Junction.Conjunction<>(instrumentation.getMethodMatcher(), not(isAbstract()));
         return agentBuilder
             .type(new AgentBuilder.RawMatcher() {
                 @Override
@@ -411,11 +413,6 @@ private static AgentBuilder getAgentBuilder(final ByteBuddy byteBuddy, final Cor
                 logger.warn("Failed to add ClassFileLocator for the agent jar. Some instrumentations may not work", e);
             }
         }
-
-        // Leave these variables here instead of invoking the config methods within the matching methods, otherwise Mockito has trouble with it
-        final List<WildcardMatcher> defaultClassesExcludedFromInstrumentation = coreConfiguration.getDefaultClassesExcludedFromInstrumentation();
-        final List<WildcardMatcher> classesExcludedFromInstrumentation = coreConfiguration.getClassesExcludedFromInstrumentation();
-
         return new AgentBuilder.Default(byteBuddy)
             .with(RedefinitionStrategy.RETRANSFORMATION)
             // when runtime attaching, only retransform up to 100 classes at once and sleep 100ms in-between as retransformation causes a stop-the-world pause
@@ -473,18 +470,8 @@ private static AgentBuilder getAgentBuilder(final ByteBuddy byteBuddy, final Cor
             .or(nameStartsWith("io.sqreen."))
             .or(nameContains("javassist"))
             .or(nameContains(".asm."))
-            .or(new ElementMatcher.Junction.AbstractBase<TypeDescription>() {
-                @Override
-                public boolean matches(TypeDescription target) {
-                    return WildcardMatcher.anyMatch(defaultClassesExcludedFromInstrumentation, target.getName()) != null;
-                }
-            })
-            .or(new ElementMatcher.Junction.AbstractBase<TypeDescription>() {
-                @Override
-                public boolean matches(TypeDescription target) {
-                    return WildcardMatcher.anyMatch(classesExcludedFromInstrumentation, target.getName()) != null;
-                }
-            })
+            .or(anyMatch(coreConfiguration.getDefaultClassesExcludedFromInstrumentation()))
+            .or(anyMatch(coreConfiguration.getClassesExcludedFromInstrumentation()))
             .disableClassFormatChanges();
     }
 

apm-agent-core/src/main/java/co/elastic/apm/agent/bci/HelperClassManager.java

@@ -209,6 +209,7 @@ public static class ForAnyClassLoader<T> extends HelperClassManager<T> {
 
         private ForAnyClassLoader(ElasticApmTracer tracer, String implementation, String... additionalHelpers) {
             super(tracer, implementation, additionalHelpers);
+            // deliberately doesn't use WeakMapSupplier as this class manages the cleanup manually
             clId2helperMap = new WeakConcurrentMap<>(false);
         }
 

apm-agent-core/src/main/java/co/elastic/apm/agent/bci/bytebuddy/CustomElementMatchers.java

@@ -24,6 +24,7 @@
  */
 package co.elastic.apm.agent.bci.bytebuddy;
 
+import co.elastic.apm.agent.collections.WeakMapSupplier;
 import co.elastic.apm.agent.matcher.AnnotationMatcher;
 import co.elastic.apm.agent.matcher.WildcardMatcher;
 import co.elastic.apm.agent.util.Version;
@@ -45,6 +46,7 @@
 import java.security.CodeSource;
 import java.security.ProtectionDomain;
 import java.util.Collection;
+import java.util.List;
 import java.util.jar.JarFile;
 import java.util.jar.Manifest;
 
@@ -83,7 +85,7 @@ public static ElementMatcher.Junction<ClassLoader> classLoaderCanLoadClass(final
         return new ElementMatcher.Junction.AbstractBase<ClassLoader>() {
 
             private final boolean loadableByBootstrapClassLoader = canLoadClass(null, className);
-            private WeakConcurrentMap<ClassLoader, Boolean> cache = new WeakConcurrentMap.WithInlinedExpunction<>();
+            private final WeakConcurrentMap<ClassLoader, Boolean> cache = WeakMapSupplier.createMap();
 
             @Override
             public boolean matches(@Nullable ClassLoader target) {
@@ -218,6 +220,20 @@ public String toString() {
         };
     }
 
+    public static ElementMatcher.Junction<NamedElement> anyMatch(final List<WildcardMatcher> matchers) {
+        return new ElementMatcher.Junction.AbstractBase<NamedElement>() {
+            @Override
+            public boolean matches(NamedElement target) {
+                return WildcardMatcher.isAnyMatch(matchers, target.getActualName());
+            }
+
+            @Override
+            public String toString() {
+                return "matches(" + matchers + ")";
+            }
+        };
+    }
+
     public static ElementMatcher.Junction<AnnotationSource> annotationMatches(final String annotationWildcard) {
         return AnnotationMatcher.annotationMatcher(annotationWildcard);
     }

apm-agent-core/src/main/java/co/elastic/apm/agent/bci/bytebuddy/SoftlyReferencingTypePoolCache.java

@@ -53,6 +53,7 @@ public class SoftlyReferencingTypePoolCache extends AgentBuilder.PoolStrategy.Wi
     /*
      * Weakly referencing ClassLoaders to avoid class loader leaks
      * Softly referencing the type pool cache so that it does not cause OOMEs
+     * deliberately doesn't use WeakMapSupplier as this class manages the cleanup manually
      */
     private final WeakConcurrentMap<ClassLoader, CacheProviderWrapper> cacheProviders =
         new WeakConcurrentMap<ClassLoader, CacheProviderWrapper>(false);

apm-agent-core/src/main/java/co/elastic/apm/agent/cache/WeakKeySoftValueLoadingCache.java

@@ -11,9 +11,9 @@
  * 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
@@ -24,6 +24,7 @@
  */
 package co.elastic.apm.agent.cache;
 
+import co.elastic.apm.agent.collections.WeakMapSupplier;
 import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -65,7 +66,7 @@ public class WeakKeySoftValueLoadingCache<K, V> {
 
     private static final Logger logger = LoggerFactory.getLogger(WeakKeySoftValueLoadingCache.class);
 
-    private final WeakConcurrentMap<K, CacheValue<K, V>> cache = new WeakConcurrentMap.WithInlinedExpunction<>();
+    private final WeakConcurrentMap<K, CacheValue<K, V>> cache = WeakMapSupplier.createMap();
     private final ValueSupplier<K, V> valueSupplier;
 
     public WeakKeySoftValueLoadingCache(ValueSupplier<K, V> valueSupplier) {

apm-agent-core/src/main/java/co/elastic/apm/agent/collections/WeakMapCleaner.java

@@ -0,0 +1,68 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2020 Elastic and contributors
+ * %%
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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.
+ * #L%
+ */
+package co.elastic.apm.agent.collections;
+
+import co.elastic.apm.agent.context.AbstractLifecycleListener;
+import co.elastic.apm.agent.impl.ElasticApmTracer;
+import co.elastic.apm.agent.util.ExecutorUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Regularly calls {@link WeakMapSupplier#expungeStaleEntries()}
+ */
+public class WeakMapCleaner extends AbstractLifecycleListener implements Runnable {
+
+    private static final Logger logger = LoggerFactory.getLogger(WeakMapCleaner.class);
+
+    private final ScheduledThreadPoolExecutor scheduler;
+
+    public WeakMapCleaner() {
+        this.scheduler = ExecutorUtils.createSingleThreadSchedulingDeamonPool("weak-map-cleaner");
+    }
+
+    @Override
+    public void start(ElasticApmTracer tracer) {
+        scheduler.scheduleWithFixedDelay(this, 1, 1, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void stop() throws Exception {
+        scheduler.shutdownNow();
+        scheduler.awaitTermination(1, TimeUnit.SECONDS);
+    }
+
+    @Override
+    public void run() {
+        try {
+            WeakMapSupplier.expungeStaleEntries();
+        } catch (Exception e) {
+            logger.error(e.getMessage(), e);
+        }
+    }
+}

apm-agent-core/src/main/java/co/elastic/apm/agent/collections/WeakMapSupplier.java

@@ -0,0 +1,54 @@
+/*-
+ * #%L
+ * Elastic APM Java agent
+ * %%
+ * Copyright (C) 2018 - 2020 Elastic and contributors
+ * %%
+ * Licensed to Elasticsearch B.V. under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch B.V. licenses this file to you 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.
+ * #L%
+ */
+package co.elastic.apm.agent.collections;
+
+import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentMap;
+import com.blogspot.mydailyjava.weaklockfree.WeakConcurrentSet;
+
+/**
+ * The canonical place to get a new instance of a {@link WeakConcurrentMap}.
+ * Do not instantiate a {@link WeakConcurrentMap} directly to benefit from the global cleanup of stale entries.
+ */
+public class WeakMapSupplier {
+    private static final WeakConcurrentSet<WeakConcurrentMap<?, ?>> registeredMaps = new WeakConcurrentSet<>(WeakConcurrentSet.Cleaner.INLINE);
+
+    public static <K, V> WeakConcurrentMap<K, V> createMap() {
+        WeakConcurrentMap<K, V> result = new WeakConcurrentMap<>(false);
+        registeredMaps.add(result);
+        return result;
+    }
+
+    /**
+     * Calls {@link WeakConcurrentMap#expungeStaleEntries()} on all registered maps,
+     * causing the entries of already collected keys to be removed.
+     * Avoids that the maps take unnecessary space for the {@link java.util.Map.Entry}, the {@link java.lang.ref.WeakReference} and the value.
+     * Failing to call this does not mean the keys cannot be collected.
+     */
+    static void expungeStaleEntries() {
+        for (WeakConcurrentMap<?, ?> weakMap : registeredMaps) {
+            weakMap.expungeStaleEntries();
+        }
+    }
+}