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();
+}