OpenTelemetry bridge

.github/dependabot.yml

@@ -32,6 +32,7 @@ updates:
       - dependency-name: "com.datastax.cassandra:cassandra-driver-core"
       - dependency-name: "com.datastax.oss:java-driver-core"
       - dependency-name: "io.micrometer:*"
+      - dependency-name: "io.opentelemetry:opentelemetry-api"
     ignore:
       - dependency-name: "net.bytebuddy:byte-buddy-agent"
         # We deliberately want to keep this older version of Byte Buddy for our runtime attach test

CHANGELOG.asciidoc

@@ -25,6 +25,7 @@ endif::[]
 
 [float]
 ===== Features
+* Add experimental OpenTelemetry API bridge - {pull}1631[#1631]
 
 [float]
 ===== Bug fixes

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

@@ -31,6 +31,7 @@
 import co.elastic.apm.agent.impl.stacktrace.StacktraceConfiguration;
 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.Span;
 import co.elastic.apm.agent.impl.transaction.TextHeaderGetter;
 import co.elastic.apm.agent.impl.transaction.TraceContext;
@@ -86,13 +87,13 @@ public class ElasticApmTracer implements Tracer {
     private final ObjectPool<ErrorCapture> errorPool;
     private final Reporter reporter;
     private final ObjectPoolFactory objectPoolFactory;
-    // Maintains a stack of all the activated spans
+    // Maintains a stack of all the activated spans/contexts
     // This way its easy to retrieve the bottom of the stack (the transaction)
     // Also, the caller does not have to keep a reference to the previously active span, as that is maintained by the stack
-    private final ThreadLocal<Deque<AbstractSpan<?>>> activeStack = new ThreadLocal<Deque<AbstractSpan<?>>>() {
+    private final ThreadLocal<Deque<ElasticContext<?>>> activeStack = new ThreadLocal<Deque<ElasticContext<?>>>() {
         @Override
-        protected Deque<AbstractSpan<?>> initialValue() {
-            return new ArrayDeque<AbstractSpan<?>>();
+        protected Deque<ElasticContext<?>> initialValue() {
+            return new ArrayDeque<ElasticContext<?>>();
         }
     };
 
@@ -255,7 +256,7 @@ private Transaction createTransaction() {
     @Override
     @Nullable
     public Transaction currentTransaction() {
-        final AbstractSpan<?> bottomOfStack = activeStack.get().peekLast();
+        final ElasticContext<?> bottomOfStack = activeStack.get().peekLast();
         return bottomOfStack != null ? bottomOfStack.getTransaction() : null;
     }
 
@@ -476,6 +477,12 @@ public ObjectPoolFactory getObjectPoolFactory() {
     @Override
     @Nullable
     public AbstractSpan<?> getActive() {
+        ElasticContext<?> active = activeStack.get().peek();
+        return active != null ? active.getSpan() : null;
+    }
+
+    @Nullable
+    public ElasticContext<?> getActiveContext() {
         return activeStack.get().peek();
     }
 
@@ -680,54 +687,105 @@ public <T> T getLifecycleListener(Class<T> listenerClass) {
         return null;
     }
 
-    public void activate(AbstractSpan<?> span) {
+    @Nullable
+    public ElasticContext<?> currentContext() {
+        return activeStack.get().peek();
+    }
+
+    public void activate(ElasticContext<?> context) {
         if (logger.isDebugEnabled()) {
-            logger.debug("Activating {} on thread {}", span, Thread.currentThread().getId());
-        }
-        span.incrementReferences();
-        List<ActivationListener> activationListeners = getActivationListeners();
-        for (int i = 0, size = activationListeners.size(); i < size; i++) {
-            try {
-                activationListeners.get(i).beforeActivate(span);
-            } catch (Error e) {
-                throw e;
-            } catch (Throwable t) {
-                logger.warn("Exception while calling {}#beforeActivate", activationListeners.get(i).getClass().getSimpleName(), t);
+            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(span);
+
+        activeStack.get().push(newContext);
+    }
+
+    public Scope activateInScope(final ElasticContext<?> context) {
+        // already in scope
+        if (getActiveContext() == context) {
+            return Scope.NoopScope.INSTANCE;
+        }
+        context.activate();
+
+        if (context instanceof Scope) {
+            // we can take shortcut and avoid creating a separate object
+            return (Scope) context;
+        }
+        return new Scope() {
+            @Override
+            public void close() {
+                context.deactivate();
+            }
+        };
     }
 
-    public void deactivate(AbstractSpan<?> span) {
+    public void deactivate(ElasticContext<?> context) {
         if (logger.isDebugEnabled()) {
-            logger.debug("Deactivating {} on thread {}", span, Thread.currentThread().getId());
+            logger.debug("Deactivating {} on thread {}", context, Thread.currentThread().getId());
         }
+        ElasticContext<?> activeContext = activeStack.get().poll();
+        AbstractSpan<?> span = context.getSpan();
+
+        if (activeContext != context && context == span) {
+            // when context has been upgraded, we need to deactivate the original span
+            activeContext = context;
+        }
+
         try {
-            final Deque<AbstractSpan<?>> stack = activeStack.get();
-            assertIsActive(span, stack.poll());
-            List<ActivationListener> activationListeners = getActivationListeners();
-            for (int i = 0, size = activationListeners.size(); i < size; i++) {
-                try {
-                    // `this` is guaranteed to not be recycled yet as the reference count is only decremented after this method has executed
-                    activationListeners.get(i).afterDeactivate(span);
-                } catch (Error e) {
-                    throw e;
-                } catch (Throwable t) {
-                    logger.warn("Exception while calling {}#afterDeactivate", activationListeners.get(i).getClass().getSimpleName(), t);
-                }
+            assertIsActive(context, activeContext);
+
+            if (null != span) {
+                triggerActivationListeners(span, false);
             }
         } finally {
-            span.decrementReferences();
+            if (null != span) {
+                span.decrementReferences();
+            }
         }
     }
 
-    private void assertIsActive(AbstractSpan<?> span, @Nullable AbstractSpan<?> currentlyActive) {
-        if (span != currentlyActive) {
-            logger.warn("Deactivating a span ({}) which is not the currently active span ({}). " +
-                "This can happen when not properly deactivating a previous span.", span, currentlyActive);
+    private void assertIsActive(ElasticContext<?> context, @Nullable ElasticContext<?> currentlyActive) {
+        if (context != currentlyActive) {
+            logger.warn("Deactivating a context ({}) which is not the currently active one ({}). " +
+                "This can happen when not properly deactivating a previous span or context.", context, currentlyActive);
 
             if (assertionsEnabled) {
-                throw new AssertionError("Deactivating a span that is not the active one");
+                throw new AssertionError("Deactivating a context that is not the active one");
+            }
+        }
+    }
+
+    private void triggerActivationListeners(AbstractSpan<?> span, boolean isActivate) {
+        List<ActivationListener> activationListeners = getActivationListeners();
+        for (int i = 0, size = activationListeners.size(); i < size; i++) {
+            ActivationListener listener = activationListeners.get(i);
+            try {
+                if (isActivate) {
+                    listener.beforeActivate(span);
+                } else {
+                    // `this` is guaranteed to not be recycled yet as the reference count is only decremented after this method has executed
+                    listener.afterDeactivate(span);
+                }
+            } catch (Error e) {
+                throw e;
+            } catch (Throwable t) {
+                logger.warn("Exception while calling {}#{}", listener.getClass().getSimpleName(), isActivate ? "beforeActivate" : "afterDeactivate", t);
             }
         }
     }

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

@@ -89,8 +89,6 @@ public StringBuilder getFull() {
 
     /**
      * Updates full URL from current state of {@literal this}. Must be called after all other Url fields are set.
-     *
-     * @return url
      */
     private void updateFull() {
         // inspired by org.apache.catalina.connector.Request.getRequestURL
@@ -262,7 +260,7 @@ public void parseAndFillFromFull() {
         }
     }
 
-    private static int normalizePort(int port, @Nullable String protocol) {
+    public static int normalizePort(int port, @Nullable String protocol) {
         int portValue = port;
         if (portValue < 0 && protocol != null) {
             // Work around java.net.URL bug

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

@@ -30,11 +30,13 @@
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 
-public abstract class AbstractSpan<T extends AbstractSpan<T>> implements Recyclable {
+public abstract class AbstractSpan<T extends AbstractSpan<T>> implements Recyclable, ElasticContext<T> {
     public static final int PRIO_USER_SUPPLIED = 1000;
     public static final int PRIO_HIGH_LEVEL_FRAMEWORK = 100;
     public static final int PRIO_METHOD_SIGNATURE = 100;
@@ -105,6 +107,11 @@ public abstract class AbstractSpan<T extends AbstractSpan<T>> implements Recycla
 
     private boolean hasCapturedExceptions;
 
+    @Nullable
+    private OTelSpanKind otelKind = null;
+
+    private final Map<String, Object> otelAttributes = new HashMap<>();
+
     public int getReferenceCount() {
         return references.get();
     }
@@ -133,9 +140,6 @@ public boolean isDiscarded() {
         return discardRequested && getTraceContext().isDiscardable();
     }
 
-    @Nullable
-    public abstract Transaction getTransaction();
-
     private static class ChildDurationTimer implements Recyclable {
 
         private AtomicInteger activeChildren = new AtomicInteger();
@@ -356,6 +360,8 @@ public void resetState() {
         outcome = null;
         userOutcome = null;
         hasCapturedExceptions = false;
+        otelKind = null;
+        otelAttributes.clear();
     }
 
     public Span createSpan() {
@@ -491,28 +497,21 @@ private boolean hasChildId(Id spanId) {
         return false;
     }
 
+    @Override
     public T activate() {
         tracer.activate(this);
         return (T) this;
     }
 
+    @Override
     public T deactivate() {
         tracer.deactivate(this);
         return (T) this;
     }
 
+    @Override
     public Scope activateInScope() {
-        // already in scope
-        if (tracer.getActive() == this) {
-            return Scope.NoopScope.INSTANCE;
-        }
-        activate();
-        return new Scope() {
-            @Override
-            public void close() {
-                deactivate();
-            }
-        };
+        return tracer.activateInScope(this);
     }
 
     /**
@@ -663,4 +662,30 @@ public T withUserOutcome(Outcome outcome) {
         return thiz();
     }
 
+    @Override
+    public ElasticContext<T> withActiveSpan(AbstractSpan<?> span) {
+        // for internal spans the active span is only stored implicitly in the stack, hence we have no requirement
+        // to have any other kind of context storage.
+        return this;
+    }
+
+    @Override
+    public AbstractSpan<?> getSpan() {
+        return this;
+    }
+
+    public T withOtelKind(OTelSpanKind kind) {
+        this.otelKind = kind;
+        return thiz();
+    }
+
+    @Nullable
+    public OTelSpanKind getOtelKind() {
+        return otelKind;
+    }
+
+    public Map<String, Object> getOtelAttributes() {
+        return otelAttributes;
+    }
+
 }

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

@@ -0,0 +1,68 @@
+/*
+ * 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.
+ */
+package co.elastic.apm.agent.impl.transaction;
+
+import co.elastic.apm.agent.impl.Scope;
+
+import javax.annotation.Nullable;
+
+public interface ElasticContext<T extends ElasticContext<T>> {
+
+    /**
+     * Makes the context active
+     *
+     * @return this
+     */
+    T activate();
+
+    /**
+     * Deactivates context
+     *
+     * @return this
+     */
+    T deactivate();
+
+    /**
+     * Activates context in a scope
+     *
+     * @return active scope that will deactivate context when closed
+     */
+    Scope activateInScope();
+
+    /**
+     * Adds a span as active within context. Might return a different context instance if required, for example
+     * when the context implementation is immutable and thus can't be modified.
+     *
+     * @param span span to add to the context
+     * @return context with activated span
+     */
+    ElasticContext<T> withActiveSpan(AbstractSpan<?> span);
+
+    /**
+     * @return the span/transaction that is associated to this context, {@literal null} if there is none
+     */
+    @Nullable
+    AbstractSpan<?> getSpan();
+
+    /**
+     * @return transaction associated to this context, {@literal null} if there is none
+     */
+    @Nullable
+    Transaction getTransaction();
+}