Universal profiling integration: Add stacktrace-IDs to transactions (#…

…3615)

CHANGELOG.asciidoc

@@ -31,6 +31,10 @@ Use subheadings with the "=====" level for adding notes for unreleased changes:
 
 === Unreleased
 
+[float]
+===== Features
+* Added support for correlating APM data with elastic universal profiling data - {pull}3615[#3615], {pull}3602[#3602], {pull}3607[#3607], {pull}3598[#3598]
+
 [float]
 ===== Bug fixes
 * Fixed edge case where inferred spans could cause cycles in the trace parent-child relationships, subsequently resulting in the UI crashing - {pull}3588[#3588]

apm-agent-core/pom.xml

@@ -103,7 +103,7 @@
         <dependency>
             <groupId>co.elastic.otel</groupId>
             <artifactId>jvmti-access</artifactId>
-            <version>0.3.0</version>
+            <version>0.3.1</version>
         </dependency>
         <!--
         We can't use caffeine due to requiring Java 7.

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

@@ -29,26 +29,26 @@ public class UniversalProfilingConfiguration extends ConfigurationOptionProvider
 
     private final ConfigurationOption<Boolean> enabled = ConfigurationOption.booleanOption()
         .key("universal_profiling_integration_enabled")
-        .tags("added[1.50.0]", "internal")
+        .tags("added[1.50.0]")
         .configurationCategory(PROFILING_CATEGORY)
         .description("If enabled, the apm agent will correlate it's transaction with the profiling data from elastic universal profiling running on the same host.")
         .buildWithDefault(false);
 
-    private final ConfigurationOption<Long> bufferSize = ConfigurationOption.longOption()
+    private final ConfigurationOption<Integer> bufferSize = ConfigurationOption.integerOption()
         .key("universal_profiling_integration_buffer_size")
-        .addValidator(isInRange(64L, Long.MAX_VALUE))
-        .tags("added[1.50.0]", "internal")
+        .addValidator(isInRange(64, Integer.MAX_VALUE))
+        .tags("added[1.50.0]")
         .configurationCategory(PROFILING_CATEGORY)
         .description("The feature needs to buffer ended local-root spans for a short duration to ensure that all of its profiling data has been received." +
                      "This configuration option configures the buffer size in number of spans. " +
                      "The higher the number of local root spans per second, the higher this buffer size should be set.\n" +
                      "The agent will log a warning if it is not capable of buffering a span due to insufficient buffer size. " +
                      "This will cause the span to be exported immediately instead with possibly incomplete profiling correlation data.")
-        .buildWithDefault(4096L);
+        .buildWithDefault(4096);
 
     private final ConfigurationOption<String> socketDir = ConfigurationOption.stringOption()
         .key("universal_profiling_integration_socket_dir")
-        .tags("added[1.50.0]", "internal")
+        .tags("added[1.50.0]")
         .configurationCategory(PROFILING_CATEGORY)
         .description("The extension needs to bind a socket to a file for communicating with the universal profiling host agent." +
                      "This configuration option can be used to change the location. " +
@@ -60,7 +60,7 @@ public boolean isEnabled() {
         return enabled.get();
     }
 
-    public long getBufferSize() {
+    public int getBufferSize() {
         return bufferSize.get();
     }
 

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

@@ -673,6 +673,10 @@ public synchronized void stop() {
     public Reporter getReporter() {
         return reporter;
     }
+
+    public UniversalProfilingIntegration getProfilingIntegration() {
+        return profilingIntegration;
+    }
 
     public Sampler getSampler() {
         return sampler;

apm-agent-core/src/main/java/co/elastic/apm/agent/report/serialize/Base64SerializationUtils.java

@@ -1,3 +1,21 @@
+/*
+ * 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.report.serialize;
 
 import com.dslplatform.json.JsonWriter;

apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/Clock.java

@@ -0,0 +1,31 @@
+/*
+ * 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.universalprofiling;
+
+public interface Clock {
+
+    Clock SYSTEM_NANOTIME = new Clock() {
+        @Override
+        public long getNanos() {
+            return System.nanoTime();
+        }
+    };
+
+    long getNanos();
+}

apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/MoveableEvent.java

@@ -0,0 +1,30 @@
+/*
+ * 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.universalprofiling;
+
+public interface MoveableEvent<SELF extends MoveableEvent<?>> {
+
+    /**
+     * Moves the content from this event into the provided other event. This event should be in a
+     * resetted state after the call.
+     */
+    void moveInto(SELF other);
+
+    void clear();
+}

apm-agent-core/src/main/java/co/elastic/apm/agent/universalprofiling/PeekingPoller.java

@@ -0,0 +1,90 @@
+/*
+ * 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.universalprofiling;
+
+import com.lmax.disruptor.EventFactory;
+import com.lmax.disruptor.EventPoller;
+
+import java.util.function.Supplier;
+
+/**
+ * Wrapper around {@link EventPoller} which allows to "peek" elements. The provided event handling
+ * callback can decide to not handle an event. In that case, the event will be provided again as
+ * first element on the next call to {@link #poll(Handler)}.
+ */
+public class PeekingPoller<Event extends MoveableEvent<Event>> {
+
+    public interface Handler<Event extends MoveableEvent<Event>> {
+
+        /**
+         * Handles an event fetched from the ring buffer.
+         *
+         * @return true, if the event was handled and shall be removed. False if the event was not
+         * handled, no further invocations of handleEvent are desired and the same event shall be
+         * provided for the next {@link PeekingPoller#poll(Handler)} call.
+         */
+        boolean handleEvent(Event e);
+    }
+
+    private final EventPoller<Event> poller;
+    private final Event peekedEvent;
+    private boolean peekedEventPopulated;
+
+    Handler<? super Event> currentHandler;
+    private final EventPoller.Handler<Event> subHandler = new EventPoller.Handler<Event>() {
+        @Override
+        public boolean onEvent(Event event, long sequence, boolean endOfBatch) throws Exception {
+            return handleEvent(event, sequence, endOfBatch);
+        }
+    };
+
+    public PeekingPoller(EventPoller<Event> wrappedPoller, EventFactory<Event> emptyEventFactory) {
+        this.poller = wrappedPoller;
+        peekedEvent = emptyEventFactory.newInstance();
+        peekedEventPopulated = false;
+    }
+
+    public synchronized void poll(Handler<? super Event> handler) throws Exception {
+        if (peekedEventPopulated) {
+            boolean handled = handler.handleEvent(peekedEvent);
+            if (!handled) {
+                return;
+            }
+            peekedEvent.clear();
+            peekedEventPopulated = false;
+        }
+        currentHandler = handler;
+        try {
+            poller.poll(subHandler);
+        } finally {
+            currentHandler = null;
+        }
+    }
+
+    private boolean handleEvent(Event event, long sequence, boolean endOfBatch) {
+        boolean handled = currentHandler.handleEvent(event);
+        if (handled) {
+            return true;
+        } else {
+            peekedEventPopulated = true;
+            event.moveInto(peekedEvent);
+            return false;
+        }
+    }
+}