Skip to content

Commit cc4d8a1

Browse files
JonasKunzSylvainJuge
JonasKunz
and
SylvainJuge
authored
Universal profiling integration: write shared memory (#3598)
 --------- Co-authored-by: SylvainJuge <[email protected]>
1 parent 35e0d6d commit cc4d8a1

File tree

10 files changed

+555
-3
lines changed

10 files changed

+555
-3
lines changed

‎apm-agent-common/src/main/java/co/elastic/apm/agent/common/util/AgentInfo.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,8 @@ public class AgentInfo {
3636
"com.blogspot.mydailyjava.weaklockfree",
3737
"com.lmax.disruptor",
3838
"com.dslplatform.json",
39-
"com.googlecode.concurrentlinkedhashmap"
39+
"com.googlecode.concurrentlinkedhashmap",
40+
"co.elastic.otel"
4041
));
4142

4243
private static final Set<String> agentRootPackages = new HashSet<>(Arrays.asList(

‎apm-agent-core/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,11 @@
100100
<artifactId>HdrHistogram</artifactId>
101101
<version>2.1.11</version>
102102
</dependency>
103+
<dependency>
104+
<groupId>co.elastic.otel</groupId>
105+
<artifactId>jvmti-access</artifactId>
106+
<version>0.3.0</version>
107+
</dependency>
103108
<!--
104109
We can't use caffeine due to requiring Java 7.
105110
As recommended by the author, we use concurrentlinkedhashmap-lru instead:

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

Lines changed: 72 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,72 @@
1+
/*
2+
* Licensed to Elasticsearch B.V. under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch B.V. licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package co.elastic.apm.agent.configuration;
20+
21+
import org.stagemonitor.configuration.ConfigurationOption;
22+
import org.stagemonitor.configuration.ConfigurationOptionProvider;
23+
24+
import static co.elastic.apm.agent.tracer.configuration.RangeValidator.isInRange;
25+
26+
public class UniversalProfilingConfiguration extends ConfigurationOptionProvider {
27+
28+
private static final String PROFILING_CATEGORY = "Profiling";
29+
30+
private final ConfigurationOption<Boolean> enabled = ConfigurationOption.booleanOption()
31+
.key("universal_profiling_integration_enabled")
32+
.tags("added[1.50.0]", "internal")
33+
.configurationCategory(PROFILING_CATEGORY)
34+
.description("If enabled, the apm agent will correlate it's transaction with the profiling data from elastic universal profiling running on the same host.")
35+
.buildWithDefault(false);
36+
37+
private final ConfigurationOption<Long> bufferSize = ConfigurationOption.longOption()
38+
.key("universal_profiling_integration_buffer_size")
39+
.addValidator(isInRange(64L, Long.MAX_VALUE))
40+
.tags("added[1.50.0]", "internal")
41+
.configurationCategory(PROFILING_CATEGORY)
42+
.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." +
43+
"This configuration option configures the buffer size in number of spans. " +
44+
"The higher the number of local root spans per second, the higher this buffer size should be set.\n" +
45+
"The agent will log a warning if it is not capable of buffering a span due to insufficient buffer size. " +
46+
"This will cause the span to be exported immediately instead with possibly incomplete profiling correlation data.")
47+
.buildWithDefault(4096L);
48+
49+
private final ConfigurationOption<String> socketDir = ConfigurationOption.stringOption()
50+
.key("universal_profiling_integration_socket_dir")
51+
.tags("added[1.50.0]", "internal")
52+
.configurationCategory(PROFILING_CATEGORY)
53+
.description("The extension needs to bind a socket to a file for communicating with the universal profiling host agent." +
54+
"This configuration option can be used to change the location. " +
55+
"Note that the total path name (including the socket) must not exceed 100 characters due to OS restrictions.\n" +
56+
"If unset, the value of the `java.io.tmpdir` system property will be used.")
57+
.build();
58+
59+
public boolean isEnabled() {
60+
return enabled.get();
61+
}
62+
63+
public long getBufferSize() {
64+
return bufferSize.get();
65+
}
66+
67+
public String getSocketDir() {
68+
String dir = socketDir.get();
69+
return dir == null || dir.isEmpty() ? System.getProperty("java.io.tmpdir") : dir;
70+
}
71+
72+
}

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

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@
7171
import co.elastic.apm.agent.tracer.dispatch.HeaderGetter;
7272
import co.elastic.apm.agent.tracer.reference.ReferenceCounted;
7373
import co.elastic.apm.agent.tracer.reference.ReferenceCountedMap;
74+
import co.elastic.apm.agent.universalprofiling.UniversalProfilingIntegration;
7475
import co.elastic.apm.agent.util.DependencyInjectingServiceLoader;
7576
import co.elastic.apm.agent.util.ExecutorUtils;
7677
import com.dslplatform.json.JsonWriter;
@@ -143,6 +144,8 @@ protected ActiveStack initialValue() {
143144
private final SpanConfiguration spanConfiguration;
144145
private final List<ActivationListener> activationListeners;
145146
private final MetricRegistry metricRegistry;
147+
148+
private final UniversalProfilingIntegration profilingIntegration;
146149
private final ScheduledThreadPoolExecutor sharedPool;
147150
private final int approximateContextSize;
148151
private Sampler sampler;
@@ -261,6 +264,7 @@ public void onChange(ConfigurationOption<?> configurationOption, Double oldValue
261264
// sets the assertionsEnabled flag to true if indeed enabled
262265
//noinspection AssertWithSideEffects
263266
assert assertionsEnabled = true;
267+
profilingIntegration = new UniversalProfilingIntegration();
264268
}
265269

266270
@Override
@@ -341,6 +345,7 @@ private void afterTransactionStart(@Nullable ClassLoader initiatingClassLoader,
341345
if (serviceInfo != null) {
342346
transaction.getTraceContext().setServiceInfo(serviceInfo.getServiceName(), serviceInfo.getServiceVersion());
343347
}
348+
profilingIntegration.afterTransactionStart(transaction);
344349
}
345350

346351
public Transaction noopTransaction() {
@@ -525,8 +530,9 @@ public void endTransaction(Transaction transaction) {
525530
if (!transaction.isNoop() &&
526531
(transaction.isSampled() || apmServerClient.supportsKeepingUnsampledTransaction())) {
527532
// we do report non-sampled transactions (without the context)
528-
reporter.report(transaction);
533+
profilingIntegration.correlateAndReport(transaction);
529534
} else {
535+
profilingIntegration.drop(transaction);
530536
transaction.decrementReferences();
531537
}
532538
}
@@ -633,6 +639,7 @@ public synchronized void stop() {
633639
logger.debug("Tracer stop stack trace: ", new Throwable("Expected - for debugging purposes"));
634640
}
635641

642+
profilingIntegration.stop();
636643
try {
637644
configurationRegistry.close();
638645
reporter.close();
@@ -738,6 +745,7 @@ private synchronized void startSync() {
738745
}
739746
apmServerClient.start();
740747
reporter.start();
748+
profilingIntegration.start(this);
741749
for (LifecycleListener lifecycleListener : lifecycleListeners) {
742750
try {
743751
lifecycleListener.start(this);

‎apm-agent-core/src/main/java/co/elastic/apm/agent/impl/metadata/SystemInfo.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -138,7 +138,7 @@ public static SystemInfo create(final @Nullable String configuredHostname, final
138138
return systemInfo.findContainerDetails();
139139
}
140140

141-
static boolean isWindows(String osName) {
141+
public static boolean isWindows(String osName) {
142142
return osName.startsWith("Windows");
143143
}
144144

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,10 @@ public int toBytes(byte[] bytes, int offset) {
8989
return offset + data.length;
9090
}
9191

92+
public void writeToBuffer(ByteBuffer buffer) {
93+
buffer.put(data);
94+
}
95+
9296
public void fromLongs(long... values) {
9397
if (values.length * Long.BYTES != data.length) {
9498
throw new IllegalArgumentException("Invalid number of long values");

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

Lines changed: 114 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
/*
2+
* Licensed to Elasticsearch B.V. under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch B.V. licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package co.elastic.apm.agent.universalprofiling;
20+
21+
import co.elastic.apm.agent.impl.transaction.AbstractSpan;
22+
import co.elastic.apm.agent.impl.transaction.Transaction;
23+
import co.elastic.apm.agent.sdk.logging.Logger;
24+
import co.elastic.apm.agent.sdk.logging.LoggerFactory;
25+
import co.elastic.otel.UniversalProfilingCorrelation;
26+
27+
import javax.annotation.Nullable;
28+
import java.nio.ByteBuffer;
29+
import java.nio.ByteOrder;
30+
import java.nio.charset.StandardCharsets;
31+
32+
public class ProfilerSharedMemoryWriter {
33+
34+
private static final Logger log = LoggerFactory.getLogger(ProfilerSharedMemoryWriter.class);
35+
36+
private static final int TLS_MINOR_VERSION_OFFSET = 0;
37+
private static final int TLS_VALID_OFFSET = 2;
38+
private static final int TLS_TRACE_PRESENT_OFFSET = 3;
39+
private static final int TLS_TRACE_FLAGS_OFFSET = 4;
40+
private static final int TLS_TRACE_ID_OFFSET = 5;
41+
private static final int TLS_SPAN_ID_OFFSET = 21;
42+
private static final int TLS_LOCAL_ROOT_SPAN_ID_OFFSET = 29;
43+
static final int TLS_STORAGE_SIZE = 37;
44+
45+
private static volatile int writeForMemoryBarrier = 0;
46+
47+
static ByteBuffer generateProcessCorrelationStorage(String serviceName, @Nullable String environment, String socketFilePath) {
48+
ByteBuffer buffer = ByteBuffer.allocateDirect(4096);
49+
buffer.order(ByteOrder.nativeOrder());
50+
buffer.position(0);
51+
52+
buffer.putChar((char) 1); // layout-minor-version
53+
writeUtf8Str(buffer, serviceName);
54+
writeUtf8Str(buffer, environment == null ? "" : environment);
55+
writeUtf8Str(buffer, socketFilePath);
56+
return buffer;
57+
}
58+
59+
private static void writeUtf8Str(ByteBuffer buffer, String str) {
60+
byte[] utf8 = str.getBytes(StandardCharsets.UTF_8);
61+
buffer.putInt(utf8.length);
62+
buffer.put(utf8);
63+
}
64+
65+
/**
66+
* This method ensures that all writes which happened prior to this method call are not moved
67+
* after the method call due to reordering.
68+
*
69+
* <p>This is realized based on the Java Memory Model guarantess for volatile variables. Relevant
70+
* resources:
71+
*
72+
* <ul>
73+
* <li><a
74+
* href="https://stackoverflow.com/questions/17108541/happens-before-relationships-with-volatile-fields-and-synchronized-blocks-in-jav">StackOverflow
75+
* topic</a>
76+
* <li><a href="https://gee.cs.oswego.edu/dl/jmm/cookbook.html">JSR Compiler Cookbook</a>
77+
* </ul>
78+
*/
79+
private static void memoryStoreStoreBarrier() {
80+
writeForMemoryBarrier = 42;
81+
}
82+
83+
static void updateThreadCorrelationStorage(@Nullable AbstractSpan<?> newSpan) {
84+
try {
85+
ByteBuffer tls = UniversalProfilingCorrelation.getCurrentThreadStorage(true, TLS_STORAGE_SIZE);
86+
// tls might be null if unsupported or something went wrong on initialization
87+
if (tls != null) {
88+
// the valid flag is used to signal the host-agent that it is reading incomplete data
89+
tls.put(TLS_VALID_OFFSET, (byte) 0);
90+
memoryStoreStoreBarrier();
91+
tls.putChar(TLS_MINOR_VERSION_OFFSET, (char) 1);
92+
93+
if (newSpan != null) {
94+
Transaction tx = newSpan.getParentTransaction();
95+
tls.put(TLS_TRACE_PRESENT_OFFSET, (byte) 1);
96+
tls.put(TLS_TRACE_FLAGS_OFFSET, newSpan.getTraceContext().getFlags());
97+
tls.position(TLS_TRACE_ID_OFFSET);
98+
newSpan.getTraceContext().getTraceId().writeToBuffer(tls);
99+
tls.position(TLS_SPAN_ID_OFFSET);
100+
newSpan.getTraceContext().getId().writeToBuffer(tls);
101+
tls.position(TLS_LOCAL_ROOT_SPAN_ID_OFFSET);
102+
tx.getTraceContext().getId().writeToBuffer(tls);
103+
} else {
104+
tls.put(TLS_TRACE_PRESENT_OFFSET, (byte) 0);
105+
}
106+
memoryStoreStoreBarrier();
107+
tls.put(TLS_VALID_OFFSET, (byte) 1);
108+
}
109+
} catch (Exception e) {
110+
log.error("Failed to write profiling correlation tls", e);
111+
}
112+
}
113+
}
114+

0 commit comments

Comments
 (0)