Fix OTel bridge behavior with multiple classloaders + improved activation (#2735)

* degug add classloader info

* test more versions + stricter tests

* minor refactor

* add external plugin test with otel

* minor cleanup

* properly size wrapper

* move wrapping to thread-local active stack

* cleanup

* improve javadoc + tune sizing

* finalize testing & bugfix

* update changelog

* fix misleading comment

* remove duplication

* clarify javadoc

* post-review changes

Author: SylvainJuge

CHANGELOG.asciidoc

@@ -52,6 +52,7 @@ version of classes that are available also in the parent (bootstrap) class loade
 * Update `async-profiler` to 1.8.8 to avoid missing symbol spam - {pull}2775[#2775]
 * Fix container ID discovery for containers managed through AWS Fargate - {pull}2772[#2772]
 * Make `traceparent` header computation thread-safe - {pull}2747[#2747]
+* Fix OTel bridge with multiple OTel APIs or external plugins - {pull}2735[#2735]
 
 [[release-notes-1.x]]
 === Java Agent version 1.x

apm-agent-core/src/main/java/co/elastic/apm/agent/configuration/CoreConfiguration.java

@@ -38,6 +38,11 @@
 import org.stagemonitor.configuration.source.ConfigurationSource;
 
 import javax.annotation.Nullable;
+import java.io.IOException;
+import java.nio.file.DirectoryStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
@@ -232,7 +237,7 @@ public class CoreConfiguration extends ConfigurationOptionProvider {
         .description("Limits the amount of spans that are recorded per transaction.\n\n" +
             "This is helpful in cases where a transaction creates a very high amount of spans (e.g. thousands of SQL queries).\n\n" +
             "Setting an upper limit will prevent overloading the agent and the APM server with too much work for such edge cases.\n\n" +
-            "A message will be logged when the max number of spans has been exceeded but only at a rate of once every " + TimeUnit.MICROSECONDS.toMinutes(Span.MAX_LOG_INTERVAL_MICRO_SECS)  + " minutes to ensure performance is not impacted.")
+            "A message will be logged when the max number of spans has been exceeded but only at a rate of once every " + TimeUnit.MICROSECONDS.toMinutes(Span.MAX_LOG_INTERVAL_MICRO_SECS) + " minutes to ensure performance is not impacted.")
         .dynamic(true)
         .buildWithDefault(500);
 
@@ -241,20 +246,20 @@ public class CoreConfiguration extends ConfigurationOptionProvider {
         .key("sanitize_field_names")
         .configurationCategory(CORE_CATEGORY)
         .description("Sometimes it is necessary to sanitize the data sent to Elastic APM,\n" +
-            "e.g. remove sensitive data.\n" +
-            "\n" +
-            "Configure a list of wildcard patterns of field names which should be sanitized.\n" +
-            "These apply for example to HTTP headers and `application/x-www-form-urlencoded` data.\n" +
-            "\n" +
-            WildcardMatcher.DOCUMENTATION + "\n" +
-            "\n" +
-            "NOTE: Data in the query string is considered non-sensitive,\n" +
-            "as sensitive information should not be sent in the query string.\n" +
-            "See https://www.owasp.org/index.php/Information_exposure_through_query_strings_in_url for more information\n" +
-            "\n" +
-            "NOTE: Review the data captured by Elastic APM carefully to make sure it does not capture sensitive information.\n" +
-            "If you do find sensitive data in the Elasticsearch index,\n" +
-            "you should add an additional entry to this list (make sure to also include the default entries)."
+                "e.g. remove sensitive data.\n" +
+                "\n" +
+                "Configure a list of wildcard patterns of field names which should be sanitized.\n" +
+                "These apply for example to HTTP headers and `application/x-www-form-urlencoded` data.\n" +
+                "\n" +
+                WildcardMatcher.DOCUMENTATION + "\n" +
+                "\n" +
+                "NOTE: Data in the query string is considered non-sensitive,\n" +
+                "as sensitive information should not be sent in the query string.\n" +
+                "See https://www.owasp.org/index.php/Information_exposure_through_query_strings_in_url for more information\n" +
+                "\n" +
+                "NOTE: Review the data captured by Elastic APM carefully to make sure it does not capture sensitive information.\n" +
+                "If you do find sensitive data in the Elasticsearch index,\n" +
+                "you should add an additional entry to this list (make sure to also include the default entries)."
             /* A disadvantage of this approach is when a user adds a custom value,
              * they don't automatically pick up new default values.
              * But the possibility to remove default values which are leading to false positive for the user
@@ -850,7 +855,7 @@ public List<WildcardMatcher> getUnnestExceptions() {
         return unnestExceptions.get();
     }
 
-    public List<WildcardMatcher> getIgnoreExceptions(){
+    public List<WildcardMatcher> getIgnoreExceptions() {
         return ignoreExceptions.get();
     }
 
@@ -967,6 +972,29 @@ public String getPluginsDir() {
         }
     }
 
+    public int getExternalPluginsCount() {
+        String pathString = getPluginsDir();
+        if (pathString == null) {
+            return 0;
+        }
+        Path pluginsDir = Paths.get(pathString);
+        if (!Files.isDirectory(pluginsDir)) {
+            return 0;
+        }
+
+        int count = 0;
+        try (DirectoryStream<Path> paths = Files.newDirectoryStream(pluginsDir)) {
+            for (Path p : paths) {
+                if (p.getFileName().endsWith(".jar")) {
+                    count++;
+                }
+            }
+        } catch (IOException e) {
+            // silently ignored
+        }
+        return count;
+    }
+
     public long getMetadataDiscoveryTimeoutMs() {
         return metadataTimeoutMs.get().getMillis();
     }

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/ElasticApmTracer.java

@@ -33,6 +33,7 @@
 import co.elastic.apm.agent.impl.transaction.AbstractSpan;
 import co.elastic.apm.agent.impl.transaction.BinaryHeaderGetter;
 import co.elastic.apm.agent.impl.transaction.ElasticContext;
+import co.elastic.apm.agent.impl.transaction.ElasticContextWrapper;
 import co.elastic.apm.agent.impl.transaction.Span;
 import co.elastic.apm.agent.impl.transaction.TextHeaderGetter;
 import co.elastic.apm.agent.impl.transaction.TraceContext;
@@ -62,6 +63,8 @@
 import java.util.Deque;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Callable;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -98,11 +101,13 @@ protected Deque<ElasticContext<?>> initialValue() {
         }
     };
 
+
     private final CoreConfiguration coreConfiguration;
     private final SpanConfiguration spanConfiguration;
     private final List<ActivationListener> activationListeners;
     private final MetricRegistry metricRegistry;
     private final ScheduledThreadPoolExecutor sharedPool;
+    private final int approximateContextSize;
     private Sampler sampler;
     boolean assertionsEnabled = false;
 
@@ -160,6 +165,11 @@ public void onChange(ConfigurationOption<?> configurationOption, Double oldValue
         this.activationListeners = DependencyInjectingServiceLoader.load(ActivationListener.class, this);
         sharedPool = ExecutorUtils.createSingleThreadSchedulingDaemonPool("shared");
 
+        // The estimated number of wrappers is linear to the number of the number of external/OTel plugins
+        // - for an internal agent context, there will be at most one wrapper per external/OTel plugin.
+        // - for a context created by an external/OTel, we have one less wrapper required
+        approximateContextSize = coreConfiguration.getExternalPluginsCount() + 1; // +1 extra is for the OTel API plugin
+
         // sets the assertionsEnabled flag to true if indeed enabled
         //noinspection AssertWithSideEffects
         assert assertionsEnabled = true;
@@ -515,15 +525,10 @@ public ObjectPoolFactory getObjectPoolFactory() {
     @Override
     @Nullable
     public AbstractSpan<?> getActive() {
-        ElasticContext<?> active = activeStack.get().peek();
+        ElasticContext<?> active = currentContext();
         return active != null ? active.getSpan() : null;
     }
 
-    @Nullable
-    public ElasticContext<?> getActiveContext() {
-        return activeStack.get().peek();
-    }
-
     @Nullable
     @Override
     public Span getActiveSpan() {
@@ -725,38 +730,70 @@ public <T> T getLifecycleListener(Class<T> listenerClass) {
         return null;
     }
 
+    /**
+     * @return the currently active context, {@literal null} if there is none.
+     */
     @Nullable
     public ElasticContext<?> currentContext() {
-        return activeStack.get().peek();
+        ElasticContext<?> current = activeStack.get().peek();
+
+        // When the active context is wrapped, the wrapper should be transparent to the caller, thus we always return
+        // the underlying wrapped context.
+        if (current instanceof ElasticContextWrapper) {
+            return ((ElasticContextWrapper<?>) current).getWrappedContext();
+        }
+        return current;
+    }
+
+    /**
+     * Lazily wraps the currently active context if required, wrapper instance is cached with wrapperClass as key.
+     * Wrapping is transparently handled by {@link #currentContext()}.
+     *
+     * @param wrapperClass wrapper type
+     * @param wrapFunction wrapper creation function
+     * @param <T>          wrapper type
+     * @return newly (or previously) created wrapper
+     */
+    public <T extends ElasticContext<T>> T wrapActiveContextIfRequired(Class<T> wrapperClass, Callable<T> wrapFunction) {
+
+        // the current context might be either a "regular" one or a "wrapped" one if it has already been wrapped
+        ElasticContext<?> current = activeStack.get().peek();
+
+        Objects.requireNonNull(current, "active context required for wrapping");
+        ElasticContextWrapper<?> wrapper;
+        if (current instanceof ElasticContextWrapper) {
+            wrapper = (ElasticContextWrapper<?>) current;
+        } else {
+            wrapper = new ElasticContextWrapper<>(approximateContextSize, current);
+        }
+        T wrapped = wrapper.wrapIfRequired(wrapperClass, wrapFunction);
+
+        // replace the currently active on the stack, however currentContext() will make sure to return the original
+        // context in order to keep wrapping transparent.
+
+        activeStack.get().remove();
+        activeStack.get().push(wrapper);
+
+        return wrapped;
     }
 
     public void activate(ElasticContext<?> context) {
         if (logger.isDebugEnabled()) {
             logger.debug("Activating {} on thread {}", context, Thread.currentThread().getId());
         }
 
-        ElasticContext<?> currentContext = currentContext();
-        ElasticContext<?> newContext = context;
-
         AbstractSpan<?> span = context.getSpan();
         if (span != null) {
             span.incrementReferences();
             triggerActivationListeners(span, true);
-        } else if(currentContext != null) {
-            // when there is no span attached to the context we are attaching to but there is one in the current
-            // context, we just propagate to the context that will be activated.
-            span = currentContext.getSpan();
-            if (span != null) {
-                newContext = context.withActiveSpan(span);
-            }
         }
 
-        activeStack.get().push(newContext);
+        activeStack.get().push(context);
     }
 
     public Scope activateInScope(final ElasticContext<?> context) {
         // already in scope
-        if (getActiveContext() == context) {
+        if (currentContext() == context) {
             return Scope.NoopScope.INSTANCE;
         }
         context.activate();
@@ -777,12 +814,14 @@ public void deactivate(ElasticContext<?> context) {
         if (logger.isDebugEnabled()) {
             logger.debug("Deactivating {} on thread {}", context, Thread.currentThread().getId());
         }
-        ElasticContext<?> activeContext = activeStack.get().poll();
+        ElasticContext<?> activeContext = currentContext();
+        activeStack.get().remove();
+
         AbstractSpan<?> span = context.getSpan();
 
-        if (activeContext != context && context == span) {
-            // when context has been upgraded, we need to deactivate the original span
-            activeContext = context;
+        if (activeContext != context && activeContext instanceof ElasticContextWrapper) {
+            // when context has been wrapped, we need to get the underlying context
+            activeContext = ((ElasticContextWrapper<?>) activeContext).getWrappedContext();
         }
 
         try {

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/AbstractSpan.java

@@ -421,9 +421,10 @@ public <H, C> boolean addSpanLink(
 
     /**
      * Adds a span link based on the tracecontext header retrieved from the provided parent.
+     *
      * @param childContextCreator the proper tracecontext inference implementation, which retrieves the header
-     * @param parent the object from which the tracecontext header is to be retrieved
-     * @param <T> the parent type - AbstractSpan, TraceContext or Tracer
+     * @param parent              the object from which the tracecontext header is to be retrieved
+     * @param <T>                 the parent type - AbstractSpan, TraceContext or Tracer
      * @return {@code true} if added, {@code false} otherwise
      */
     public <T> boolean addSpanLink(TraceContext.ChildContextCreator<T> childContextCreator, T parent) {

apm-agent-core/src/main/java/co/elastic/apm/agent/impl/transaction/ElasticContext.java

@@ -65,4 +65,5 @@ public interface ElasticContext<T extends ElasticContext<T>> {
      */
     @Nullable
     Transaction getTransaction();
+
 }