From 510900b5b5fe639239e1637ba15ebee978d0375f Mon Sep 17 00:00:00 2001 From: Jaroslav Bachorik Date: Wed, 14 Oct 2026 23:45:23 +0200 Subject: [PATCH] feat(profiling): Integrate OTLP profiles export into the agent Wire the converter into the profiler: OtlpProfileUploader consumes reference-counted RecordingData via a LightweightOtlpEncoder and ships profiles through the otlp-exporter sender, with conversion/export telemetry reported through OtlpTelemetry. Configuration surface: DD_PROFILING_OTLP_* and OTEL_EXPORTER_OTLP_PROFILES_* (endpoint, headers, protocol, compression, timeout) resolved through OtelEnvironmentConfigSource, registered in supported-configurations.json and ConfigDefaults. Included in the agent shadow jar. --- .../profiling-uploader/build.gradle | 8 +- .../uploader/LightweightOtlpEncoder.java | 216 ++++++++++ .../uploader/OtlpProfileUploader.java | 407 ++++++++++++++++++ .../uploader/OtlpProfilesSenderFactory.java | 46 ++ .../uploader/LightweightOtlpEncoderTest.java | 180 ++++++++ .../uploader/OtlpProfileUploaderTest.java | 243 +++++++++++ .../profiling/agent/ProfilingAgent.java | 67 ++- .../agent/OtlpUploadListenerTest.java | 99 +++++ dd-trace-api/build.gradle.kts | 1 + .../datadog/trace/api/ConfigDefaults.java | 7 + .../datadog/trace/api/config/OtlpConfig.java | 7 + .../trace/api/config/ProfilingConfig.java | 17 + .../main/java/datadog/trace/api/Config.java | 67 +++ .../trace/api/telemetry/OtlpTelemetry.java | 76 +++- .../api/telemetry/OtlpTelemetryTest.java | 41 ++ ...elProfilesEnvironmentConfigSourceTest.java | 116 +++++ metadata/supported-configurations.json | 98 +++++ .../provider/OtelEnvironmentConfigSource.java | 58 ++- 18 files changed, 1740 insertions(+), 14 deletions(-) create mode 100644 dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/LightweightOtlpEncoder.java create mode 100644 dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfileUploader.java create mode 100644 dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfilesSenderFactory.java create mode 100644 dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/LightweightOtlpEncoderTest.java create mode 100644 dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/OtlpProfileUploaderTest.java create mode 100644 dd-java-agent/agent-profiling/src/test/java/com/datadog/profiling/agent/OtlpUploadListenerTest.java create mode 100644 internal-api/src/test/java/datadog/trace/bootstrap/config/provider/OtelProfilesEnvironmentConfigSourceTest.java diff --git a/dd-java-agent/agent-profiling/profiling-uploader/build.gradle b/dd-java-agent/agent-profiling/profiling-uploader/build.gradle index 39c4737cfd8..2faedc6c4b3 100644 --- a/dd-java-agent/agent-profiling/profiling-uploader/build.gradle +++ b/dd-java-agent/agent-profiling/profiling-uploader/build.gradle @@ -18,12 +18,17 @@ excludedClassesCoverage += [ 'com.datadog.profiling.uploader.ProfileUploader.1', // When running on Java 8 without `jfr`, no code is executed 'com.datadog.profiling.uploader.util.JfrCliHelper', - 'com.datadog.profiling.uploader.util.JfrCliHelper.Event' + 'com.datadog.profiling.uploader.util.JfrCliHelper.Event', + // Pre-existing borderline coverage — HTTP/upload plumbing, hard to unit test + 'com.datadog.profiling.uploader.CompressingRequestBody', + 'com.datadog.profiling.uploader.ProfileUploader', + 'com.datadog.profiling.uploader.ProfileUploader.RecordingDataAdapter', ] dependencies { implementation libs.slf4j implementation project(':communication') + implementation project(':communication:otlp-exporter') implementation project(':internal-api') implementation project(':utils:container-utils') implementation project(':utils:logging-utils') @@ -31,6 +36,7 @@ dependencies { implementation project(':utils:version-utils') implementation project(':dd-java-agent:agent-profiling:profiling-controller') + implementation project(':dd-java-agent:agent-profiling:profiling-otel') implementation libs.datadog.okhttp implementation libs.lz4 diff --git a/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/LightweightOtlpEncoder.java b/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/LightweightOtlpEncoder.java new file mode 100644 index 00000000000..9ac9da7e626 --- /dev/null +++ b/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/LightweightOtlpEncoder.java @@ -0,0 +1,216 @@ +package com.datadog.profiling.uploader; + +import com.datadog.profiling.otel.proto.OtlpProtoFields; +import com.datadog.profiling.otel.proto.OtlpResourceAttributes; +import com.datadog.profiling.otel.proto.ProtobufEncoder; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.channels.FileChannel; +import java.nio.file.Path; +import java.nio.file.StandardOpenOption; +import java.time.Instant; +import java.util.Map; +import java.util.concurrent.ThreadLocalRandom; + +/** + * Encodes a minimal OTLP ProfilesData message with no sample conversion — just metadata and the raw + * JFR recording embedded as the {@code original_payload} blob. This is orders of magnitude faster + * than full JFR→OTLP conversion since it skips JFR parsing, event processing, and dictionary + * building entirely. + * + *

All nested message lengths are computed up front from the recording size, so the message is + * written in one pass into a single buffer and the JFR bytes are read from the file directly into + * it. The buffer is reused across calls; it holds the whole message, so it grows to fit the largest + * recording, and once larger than {@code MAX_RETAINED_BUFFER} it is shrunk on the next smaller + * encode. Not thread-safe: the buffer returned by {@link #encode} is only valid until the next + * call. + */ +final class LightweightOtlpEncoder { + + // string_table layout: index 0 is the null/unset sentinel required by the OTLP spec, + // indices 1 and 2 back the sample_type/period_type ValueType below + private static final int TYPE_STRINDEX_SAMPLES = 1; + private static final int UNIT_STRINDEX_COUNT = 2; + // buffer growth granularity — bounds the over-allocation while avoiding a reallocation for + // every slightly larger recording + private static final int BUFFER_GROWTH_ALIGNMENT = 1024 * 1024; + // a buffer above this size is reallocated to fit a smaller message, so a single large recording + // does not pin its buffer for the rest of the agent's lifetime + private static final int MAX_RETAINED_BUFFER = 8 * BUFFER_GROWTH_ALIGNMENT; + + private static final byte[] DICTIONARY_FIELD = encodeDictionaryField(); + + // the ResourceProfiles.resource field; resource attributes are fixed for the uploader lifetime + private final byte[] resourceField; + // the Profile fields preceding original_payload + private final ProtobufEncoder profileHeader = new ProtobufEncoder(128); + private final byte[] profileId = new byte[16]; + private byte[] buffer = new byte[0]; + + LightweightOtlpEncoder(Map resourceAttributes) { + ProtobufEncoder encoder = new ProtobufEncoder(512); + OtlpResourceAttributes.writeResource(encoder, resourceAttributes); + this.resourceField = encoder.toByteArray(); + } + + ByteBuffer encode(Path jfrFile, Instant start, Instant end) throws IOException { + long startTimeNanos = start.getEpochSecond() * 1_000_000_000L + start.getNano(); + long endTimeNanos = end.getEpochSecond() * 1_000_000_000L + end.getNano(); + profileHeader.reset(); + encodeProfileHeader(startTimeNanos, endTimeNanos); + int headerSize = profileHeader.size(); + + try (FileChannel channel = FileChannel.open(jfrFile, StandardOpenOption.READ)) { + long jfrSize = channel.size(); + long profileSize = + headerSize + fieldPrefixSize(OtlpProtoFields.Profile.ORIGINAL_PAYLOAD, jfrSize) + jfrSize; + long scopeProfilesSize = + fieldPrefixSize(OtlpProtoFields.ScopeProfiles.PROFILES, profileSize) + profileSize; + long resourceProfilesSize = + resourceField.length + + fieldPrefixSize(OtlpProtoFields.ResourceProfiles.SCOPE_PROFILES, scopeProfilesSize) + + scopeProfilesSize; + long totalSize = + fieldPrefixSize(OtlpProtoFields.ProfilesData.RESOURCE_PROFILES, resourceProfilesSize) + + resourceProfilesSize + + DICTIONARY_FIELD.length; + ensureCapacity(totalSize); + + int pos = + writeFieldPrefix( + buffer, 0, OtlpProtoFields.ProfilesData.RESOURCE_PROFILES, resourceProfilesSize); + System.arraycopy(resourceField, 0, buffer, pos, resourceField.length); + pos += resourceField.length; + pos = + writeFieldPrefix( + buffer, pos, OtlpProtoFields.ResourceProfiles.SCOPE_PROFILES, scopeProfilesSize); + pos = writeFieldPrefix(buffer, pos, OtlpProtoFields.ScopeProfiles.PROFILES, profileSize); + profileHeader.toByteBuffer().get(buffer, pos, headerSize); + pos += headerSize; + pos = writeFieldPrefix(buffer, pos, OtlpProtoFields.Profile.ORIGINAL_PAYLOAD, jfrSize); + + // the length prefixes above are derived from jfrSize, so exactly that many bytes are read; + // a recording truncated concurrently fails the export instead of emitting a malformed message + ByteBuffer target = ByteBuffer.wrap(buffer, pos, (int) jfrSize); + while (target.hasRemaining()) { + if (channel.read(target) < 0) { + throw new IOException("JFR recording shrank while being encoded: " + jfrFile); + } + } + pos += (int) jfrSize; + + System.arraycopy(DICTIONARY_FIELD, 0, buffer, pos, DICTIONARY_FIELD.length); + pos += DICTIONARY_FIELD.length; + return ByteBuffer.wrap(buffer, 0, pos); + } + } + + private void encodeProfileHeader(long startTimeNanos, long endTimeNanos) { + // Field 1: sample_type / Field 5: period_type (count-based samples) + profileHeader.writeNestedMessage( + OtlpProtoFields.Profile.SAMPLE_TYPE, + typeEncoder -> encodeValueType(typeEncoder, TYPE_STRINDEX_SAMPLES, UNIT_STRINDEX_COUNT)); + profileHeader.writeNestedMessage( + OtlpProtoFields.Profile.PERIOD_TYPE, + typeEncoder -> encodeValueType(typeEncoder, TYPE_STRINDEX_SAMPLES, UNIT_STRINDEX_COUNT)); + + profileHeader.writeFixed64Field(OtlpProtoFields.Profile.TIME_UNIX_NANO, startTimeNanos); + // clamped so a reversed window cannot underflow + profileHeader.writeVarintField( + OtlpProtoFields.Profile.DURATION_NANO, Math.max(0, endTimeNanos - startTimeNanos)); + profileHeader.writeVarintField(OtlpProtoFields.Profile.PERIOD, 1); + fillProfileId(); + profileHeader.writeBytesField(OtlpProtoFields.Profile.PROFILE_ID, profileId); + profileHeader.writeStringField(OtlpProtoFields.Profile.ORIGINAL_PAYLOAD_FORMAT, "jfr"); + } + + private void ensureCapacity(long size) throws IOException { + if (size > Integer.MAX_VALUE - BUFFER_GROWTH_ALIGNMENT) { + throw new IOException("OTLP profile message too large: " + size + " bytes"); + } + long aligned = + (size + BUFFER_GROWTH_ALIGNMENT - 1) / BUFFER_GROWTH_ALIGNMENT * BUFFER_GROWTH_ALIGNMENT; + if (size > buffer.length || (buffer.length > MAX_RETAINED_BUFFER && aligned < buffer.length)) { + // every encode rewrites the whole message, so the old content is not copied over + buffer = new byte[(int) aligned]; + } + } + + private static long fieldPrefixSize(int fieldNumber, long length) { + return varintSize((fieldNumber << 3) | ProtobufEncoder.WIRETYPE_LENGTH_DELIMITED) + + varintSize(length); + } + + private static int writeFieldPrefix(byte[] dest, int pos, int fieldNumber, long length) { + pos = writeVarint(dest, pos, (fieldNumber << 3) | ProtobufEncoder.WIRETYPE_LENGTH_DELIMITED); + return writeVarint(dest, pos, length); + } + + private static int varintSize(long value) { + int size = 1; + while ((value & ~0x7FL) != 0) { + value >>>= 7; + size++; + } + return size; + } + + private static int writeVarint(byte[] dest, int pos, long value) { + while ((value & ~0x7FL) != 0) { + dest[pos++] = (byte) ((value & 0x7F) | 0x80); + value >>>= 7; + } + dest[pos++] = (byte) value; + return pos; + } + + private static void encodeValueType(ProtobufEncoder encoder, int typeIndex, int unitIndex) { + encoder.writeVarintField(OtlpProtoFields.ValueType.TYPE_STRINDEX, typeIndex); + encoder.writeVarintField(OtlpProtoFields.ValueType.UNIT_STRINDEX, unitIndex); + } + + // minimal dictionary: the zero-value index-0 entry the OTLP spec requires in every table, plus + // the type/unit labels referenced by sample_type/period_type in string_table + private static byte[] encodeDictionaryField() { + ProtobufEncoder encoder = new ProtobufEncoder(64); + encoder.writeNestedMessage( + OtlpProtoFields.ProfilesData.DICTIONARY, + dictionaryEncoder -> { + writeEmptyEntry(dictionaryEncoder, OtlpProtoFields.ProfilesDictionary.MAPPING_TABLE); + writeEmptyEntry(dictionaryEncoder, OtlpProtoFields.ProfilesDictionary.LOCATION_TABLE); + writeEmptyEntry(dictionaryEncoder, OtlpProtoFields.ProfilesDictionary.FUNCTION_TABLE); + writeEmptyEntry(dictionaryEncoder, OtlpProtoFields.ProfilesDictionary.LINK_TABLE); + dictionaryEncoder.writeTag( + OtlpProtoFields.ProfilesDictionary.STRING_TABLE, + ProtobufEncoder.WIRETYPE_LENGTH_DELIMITED); + dictionaryEncoder.writeString(""); // index 0 (null/unset sentinel) + dictionaryEncoder.writeTag( + OtlpProtoFields.ProfilesDictionary.STRING_TABLE, + ProtobufEncoder.WIRETYPE_LENGTH_DELIMITED); + dictionaryEncoder.writeString("samples"); + dictionaryEncoder.writeTag( + OtlpProtoFields.ProfilesDictionary.STRING_TABLE, + ProtobufEncoder.WIRETYPE_LENGTH_DELIMITED); + dictionaryEncoder.writeString("count"); + writeEmptyEntry(dictionaryEncoder, OtlpProtoFields.ProfilesDictionary.ATTRIBUTE_TABLE); + writeEmptyEntry(dictionaryEncoder, OtlpProtoFields.ProfilesDictionary.STACK_TABLE); + }); + return encoder.toByteArray(); + } + + private static void writeEmptyEntry(ProtobufEncoder encoder, int tableField) { + encoder.writeTag(tableField, ProtobufEncoder.WIRETYPE_LENGTH_DELIMITED); + encoder.writeVarint(0); + } + + private void fillProfileId() { + // ThreadLocalRandom over UUID.randomUUID (SecureRandom) — uniqueness suffices, and + // SecureRandom can block on entropy + long msb = ThreadLocalRandom.current().nextLong(); + long lsb = ThreadLocalRandom.current().nextLong(); + for (int i = 0; i < 8; i++) { + profileId[i] = (byte) ((msb >> (56 - i * 8)) & 0xFF); + profileId[i + 8] = (byte) ((lsb >> (56 - i * 8)) & 0xFF); + } + } +} diff --git a/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfileUploader.java b/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfileUploader.java new file mode 100644 index 00000000000..c02d9163e00 --- /dev/null +++ b/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfileUploader.java @@ -0,0 +1,407 @@ +/* + * Copyright 2026 Datadog + * + * Licensed 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 com.datadog.profiling.uploader; + +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED_DEFAULT; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_MODE; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_MODE_DEFAULT; +import static datadog.trace.api.telemetry.LogCollector.SEND_TELEMETRY; + +import com.datadog.profiling.otel.JfrToOtlpConverter; +import datadog.communication.otlp.OtlpCanonicalResourceAttributes; +import datadog.communication.otlp.OtlpPayload; +import datadog.communication.otlp.OtlpResponse; +import datadog.communication.otlp.OtlpSender; +import datadog.trace.api.Config; +import datadog.trace.api.config.ProfilingConfig; +import datadog.trace.api.profiling.RecordingData; +import datadog.trace.api.profiling.RecordingDataListener; +import datadog.trace.api.profiling.RecordingType; +import datadog.trace.api.telemetry.OtlpTelemetry; +import datadog.trace.bootstrap.config.provider.ConfigProvider; +import datadog.trace.util.AgentThreadFactory; +import datadog.trace.util.TempLocationManager; +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.time.Instant; +import java.util.Collections; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantLock; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public final class OtlpProfileUploader implements RecordingDataListener { + + private static final Logger log = LoggerFactory.getLogger(OtlpProfileUploader.class); + private static final int TERMINATION_TIMEOUT_SEC = 5; + // cap on the raw JFR recording embedded as original_payload in FULL mode — beyond this the + // converted profile is sent without the payload instead of materializing an unbounded blob + private static final long MAX_ORIGINAL_PAYLOAD_BYTES = 64L * 1024 * 1024; + + // canonical resource attribute keys emitted from fixed configuration; a configured profiling + // or global tag with one of these (lower-cased) keys is dropped so it cannot shadow the + // canonical value — mirrors the tracer's OTLP traces export filtering + private static final Set CANONICAL_RESOURCE_KEYS = + new HashSet<>( + java.util.Arrays.asList( + "service", + "env", + "version", + "service.name", + "deployment.environment.name", + "service.version", + "host.name", + "telemetry.sdk.name", + "telemetry.sdk.version", + "telemetry.sdk.language")); + + private final OtlpSender sender; + private final ExecutorService executor; + private final boolean enabled; + private final int terminationTimeout; + private final ProfilingConfig.OtlpMode mode; + private final Map resourceAttributes; + // null unless mode is LIGHT; its reusable buffer is guarded by exportLock + private final LightweightOtlpEncoder lightweightEncoder; + // serializes conversion and send: at most one recording is converted and held in memory at a + // time, and the LIGHT encoder buffer is not overwritten while a send is still reading it + private final ReentrantLock exportLock = new ReentrantLock(); + + public OtlpProfileUploader(final Config config, final ConfigProvider configProvider) { + this(config, configProvider, TERMINATION_TIMEOUT_SEC); + } + + OtlpProfileUploader( + final Config config, final ConfigProvider configProvider, int terminationTimeout) { + boolean enabledConfig = + configProvider.getBoolean(PROFILING_OTLP_ENABLED, PROFILING_OTLP_ENABLED_DEFAULT); + this.terminationTimeout = terminationTimeout; + // the sender owns OkHttp clients and connection pools — only build it when enabled. + // A bad optional-feature configuration must not abort profiling: degrade to disabled + // and keep the classic JFR upload. + OtlpSender createdSender = null; + if (enabledConfig) { + try { + createdSender = createSender(config); + } catch (Exception e) { + log.error("Failed to create OTLP profiles sender; OTLP profile export disabled", e); + } + } + this.enabled = enabledConfig && createdSender != null; + this.sender = createdSender; + if (enabled) { + log.warn( + SEND_TELEMETRY, + "OTLP profiles export is enabled. The OTLP profiles protocol is at Development maturity" + + " (profiles specification: Alpha) and may change incompatibly; do not use it in" + + " production."); + } + this.mode = + configProvider.getEnum( + PROFILING_OTLP_MODE, ProfilingConfig.OtlpMode.class, PROFILING_OTLP_MODE_DEFAULT); + this.resourceAttributes = buildResourceAttributes(config); + this.lightweightEncoder = + mode == ProfilingConfig.OtlpMode.LIGHT + ? new LightweightOtlpEncoder(resourceAttributes) + : null; + // a single worker with no queue: a recording that arrives while the previous export is still + // running is dropped instead of being converted and buffered alongside it + this.executor = + new ThreadPoolExecutor( + 0, + 1, + 60L, + TimeUnit.SECONDS, + new SynchronousQueue<>(), + new AgentThreadFactory(AgentThreadFactory.AgentThread.PROFILER_HTTP_DISPATCHER), + new ThreadPoolExecutor.AbortPolicy()); + log.debug( + "OTLP profile uploader initialized: endpoint={}, protocol={}", + config.getOtlpProfilesEndpoint(), + config.getOtlpProfilesProtocol()); + } + + private static OtlpSender createSender(Config config) { + OtlpSender created = OtlpProfilesSenderFactory.create(config); + if (created == null) { + throw new IllegalStateException( + "Unsupported OTLP profiles protocol: " + config.getOtlpProfilesProtocol()); + } + return created; + } + + @Override + public void onNewData(RecordingType type, RecordingData data, boolean handleSynchronously) { + upload(type, data, handleSynchronously, null); + } + + public void upload(RecordingType type, RecordingData data, boolean sync, Runnable onCompletion) { + if (!enabled) { + data.release(); + if (onCompletion != null) { + onCompletion.run(); + } + return; + } + if (sync) { + // the sync path runs on shutdown/snapshot; bound the wait behind an in-flight async export + // (which may be retrying) instead of blocking for its full duration + boolean locked = false; + try { + locked = exportLock.tryLock(terminationTimeout, TimeUnit.SECONDS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + if (!locked) { + skipExport(data, onCompletion); + return; + } + try { + // reentrant: export acquires exportLock again + export(data, onCompletion); + } finally { + exportLock.unlock(); + } + return; + } + try { + executor.execute(() -> export(data, onCompletion)); + } catch (RejectedExecutionException e) { + skipExport(data, onCompletion); + } + } + + private static void skipExport(RecordingData data, Runnable onCompletion) { + log.warn("OTLP profile export skipped: the previous export is still in progress"); + // the export never started, but the attempt happened — count it so failures cannot exceed + // attempts under backpressure + OtlpTelemetry.getInstance().onProfilesExportAttempt(); + OtlpTelemetry.getInstance().onProfilesExportComplete(false); + data.release(); + if (onCompletion != null) { + onCompletion.run(); + } + } + + private void export(RecordingData data, Runnable onCompletion) { + exportLock.lock(); + try { + try { + OtlpPayload payload; + long conversionNanos; + try { + long conversionStartNanos = System.nanoTime(); + payload = convertToOtlp(data); + conversionNanos = System.nanoTime() - conversionStartNanos; + } finally { + // the payload is fully materialized in memory, so the recording is no longer needed + data.release(); + } + // LIGHT mode does not convert the recording (it exports the raw JFR); recording its + // encoding time would skew the conversion timing telemetry + if (mode != ProfilingConfig.OtlpMode.LIGHT) { + OtlpTelemetry.getInstance().onProfilesConversion(conversionNanos); + } + if (payload == null) { + // skipped by the LIGHT-mode size cap — nothing to export + return; + } + log.debug( + "JFR to OTLP conversion took {} ms (mode={}, bytes={})", + TimeUnit.NANOSECONDS.toMillis(conversionNanos), + mode, + payload.getContentLength()); + send(payload); + } catch (Exception | LinkageError e) { + // not rethrown so that the classic JFR upload continues independently; + // LinkageError covers JVMs where the jafar parser classes cannot link + log.error("Failed to upload OTLP profile", e); + } finally { + if (onCompletion != null) { + onCompletion.run(); + } + } + } finally { + exportLock.unlock(); + } + } + + private void send(OtlpPayload payload) { + try { + OtlpTelemetry.getInstance().onProfilesExportAttempt(); + OtlpResponse response = sender.send(payload); + OtlpTelemetry.getInstance().onProfilesExportComplete(response.success()); + if (!response.success()) { + log.warn( + "OTLP profile upload failed: status={}", + response.status().isPresent() ? response.status().getAsInt() : "unknown"); + } + } catch (Exception e) { + log.error("OTLP profile upload failed", e); + OtlpTelemetry.getInstance().onProfilesExportComplete(false); + } + } + + // canonical attributes are shared with the tracer's OTLP traces export + private static Map buildResourceAttributes(Config config) { + Map attributes = new LinkedHashMap<>(); + OtlpCanonicalResourceAttributes.visit(config, attributes::put); + + // merge the user-configured tags (getMergedProfilingTags: global + profiling + runtime + + // host tags, the same set the classic uploader sends); keys already emitted as canonical + // attributes above are skipped + config + .getMergedProfilingTags() + .forEach( + (key, value) -> { + if (value == null + || value.isEmpty() + || attributes.containsKey(key) + || CANONICAL_RESOURCE_KEYS.contains(key.toLowerCase(Locale.ROOT))) { + return; + } + attributes.put(key, value); + }); + return Collections.unmodifiableMap(attributes); + } + + private OtlpPayload convertToOtlp(RecordingData data) throws IOException { + if (mode == ProfilingConfig.OtlpMode.LIGHT) { + return convertLightweight(data); + } + + JfrToOtlpConverter converter = new JfrToOtlpConverter(); + boolean includePayload = mode == ProfilingConfig.OtlpMode.FULL; + converter.setResourceAttributes(resourceAttributes); + + Path jfrFile = data.getPath(); + if (jfrFile != null) { + applyPayloadCap(converter, jfrFile, includePayload); + converter.addFile(jfrFile, data.getStart(), data.getEnd()); + return new OtlpPayload( + ByteBuffer.wrap(converter.convert(JfrToOtlpConverter.Kind.PROTO)), + OtlpPayload.PROTOBUF_CONTENT_TYPE); + } + + Path tempDir = TempLocationManager.getInstance().getTempDir(); + Path temp = Files.createTempFile(tempDir, "dd-otlp-", ".jfr"); + // the copy, the conversion, and the temp-file cleanup must share one try/finally — + // a failure while streaming would leak the temp file with the full JFR recording + try { + try (InputStream stream = data.getStream()) { + Files.copy(stream, temp, StandardCopyOption.REPLACE_EXISTING); + } + applyPayloadCap(converter, temp, includePayload); + converter.addFile(temp, data.getStart(), data.getEnd()); + return new OtlpPayload( + ByteBuffer.wrap(converter.convert(JfrToOtlpConverter.Kind.PROTO)), + OtlpPayload.PROTOBUF_CONTENT_TYPE); + } finally { + Files.deleteIfExists(temp); + } + } + + // the converted profile plus the embedded raw recording is buffered as one heap byte[]; + // cap the payload so an oversized recording is exported converted-only + private static void applyPayloadCap(JfrToOtlpConverter converter, Path jfrFile, boolean wanted) { + if (!wanted) { + return; + } + try { + long size = Files.size(jfrFile); + if (size > MAX_ORIGINAL_PAYLOAD_BYTES) { + log.warn( + "JFR recording of {} bytes exceeds the original_payload cap of {} bytes; " + + "exporting the converted profile without the original payload", + size, + MAX_ORIGINAL_PAYLOAD_BYTES); + return; + } + } catch (IOException e) { + log.warn("Could not stat JFR recording; exporting without the original payload", e); + return; + } + converter.setIncludeOriginalPayload(true); + } + + private OtlpPayload convertLightweight(RecordingData data) throws IOException { + Path jfrFile = data.getPath(); + if (jfrFile != null) { + return encodeLightweight(jfrFile, data.getStart(), data.getEnd()); + } + + // Fallback: save stream to temp file, then encode; copy, encode, and cleanup share one + // try/finally so a failed copy cannot leak the temp file. + Path tempDir = TempLocationManager.getInstance().getTempDir(); + Path temp = Files.createTempFile(tempDir, "dd-otlp-", ".jfr"); + try { + try (InputStream stream = data.getStream()) { + Files.copy(stream, temp, StandardCopyOption.REPLACE_EXISTING); + } + return encodeLightweight(temp, data.getStart(), data.getEnd()); + } finally { + Files.deleteIfExists(temp); + } + } + + // LIGHT mode exports the raw recording as the profile content, so the recording size bounds + // the payload — recordings above the cap are skipped instead of buffering an unbounded blob + private OtlpPayload encodeLightweight(Path jfrFile, Instant start, Instant end) + throws IOException { + long jfrSize = Files.size(jfrFile); + if (jfrSize > MAX_ORIGINAL_PAYLOAD_BYTES) { + log.warn( + "JFR recording of {} bytes exceeds the {} byte LIGHT-mode export cap; " + + "skipping OTLP profile export", + jfrSize, + MAX_ORIGINAL_PAYLOAD_BYTES); + return null; + } + ByteBuffer encoded = lightweightEncoder.encode(jfrFile, start, end); + return new OtlpPayload(encoded, OtlpPayload.PROTOBUF_CONTENT_TYPE); + } + + public void shutdown() { + log.debug("Shutting down OTLP profile uploader"); + executor.shutdown(); + try { + if (!executor.awaitTermination(terminationTimeout, TimeUnit.SECONDS)) { + log.warn("OTLP uploader executor did not terminate in {} seconds", terminationTimeout); + executor.shutdownNow(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + executor.shutdownNow(); + } + if (sender != null) { + sender.shutdown(); + } + } +} diff --git a/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfilesSenderFactory.java b/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfilesSenderFactory.java new file mode 100644 index 00000000000..e4334470872 --- /dev/null +++ b/dd-java-agent/agent-profiling/profiling-uploader/src/main/java/com/datadog/profiling/uploader/OtlpProfilesSenderFactory.java @@ -0,0 +1,46 @@ +package com.datadog.profiling.uploader; + +import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_GRPC_PROFILES_ENDPOINT; +import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT; + +import datadog.communication.otlp.OtlpGrpcSender; +import datadog.communication.otlp.OtlpHttpSender; +import datadog.communication.otlp.OtlpSender; +import datadog.trace.api.Config; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +// mirrors OtlpMetricsSenderFactory and OtlpLogsService +final class OtlpProfilesSenderFactory { + private static final Logger log = LoggerFactory.getLogger(OtlpProfilesSenderFactory.class); + + private OtlpProfilesSenderFactory() {} + + static OtlpSender create(Config config) { + switch (config.getOtlpProfilesProtocol()) { + case GRPC: + return new OtlpGrpcSender( + config.getOtlpProfilesEndpoint(), + "/" + DEFAULT_OTLP_GRPC_PROFILES_ENDPOINT, + config.getOtlpProfilesHeaders(), + config.getOtlpProfilesTimeout(), + config.getOtlpProfilesCompression()); + case HTTP_PROTOBUF: + return new OtlpHttpSender( + config.getOtlpProfilesEndpoint(), + "/" + DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT, + config.getOtlpProfilesHeaders(), + config.getOtlpProfilesTimeout(), + config.getOtlpProfilesCompression()); + case HTTP_JSON: + // fail fast like the default branch — a silent downgrade ships payloads with a content + // type and serialization the operator did not request + throw new IllegalStateException( + "Unsupported OTLP profiles protocol: " + config.getOtlpProfilesProtocol()); + default: + // fail fast instead of returning a sender that NPEs at first upload + throw new IllegalStateException( + "Unsupported OTLP profiles protocol: " + config.getOtlpProfilesProtocol()); + } + } +} diff --git a/dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/LightweightOtlpEncoderTest.java b/dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/LightweightOtlpEncoderTest.java new file mode 100644 index 00000000000..22179a7e9ff --- /dev/null +++ b/dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/LightweightOtlpEncoderTest.java @@ -0,0 +1,180 @@ +package com.datadog.profiling.uploader; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNotSame; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.datadog.profiling.otel.proto.OtlpProtoFields; +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.nio.file.Path; +import java.time.Instant; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import java.util.Random; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +class LightweightOtlpEncoderTest { + + private static final Instant START = Instant.ofEpochSecond(1000); + private static final Instant END = Instant.ofEpochSecond(1060); + + @TempDir Path tempDir; + + private final LightweightOtlpEncoder encoder = + new LightweightOtlpEncoder(Collections.singletonMap("service.name", "test-service")); + + @Test + void embedsRecordingInWellFormedMessage() throws IOException { + // 300 bytes needs a two-byte length varint at every nesting level + byte[] jfr = randomBytes(300); + + assertPayload(encoder.encode(write("a.jfr", jfr), START, END), jfr); + } + + @Test + void reusesBufferForSmallerRecording() throws IOException { + byte[] large = randomBytes(200_000); + byte[] small = randomBytes(1_000); + + ByteBuffer first = encoder.encode(write("large.jfr", large), START, END); + assertPayload(first, large); + ByteBuffer second = encoder.encode(write("small.jfr", small), START, END); + assertPayload(second, small); + + assertSame(first.array(), second.array()); + } + + @Test + void growsBufferForLargerRecording() throws IOException { + byte[] small = randomBytes(1_000); + byte[] large = randomBytes(2 * 1024 * 1024); + + ByteBuffer first = encoder.encode(write("small.jfr", small), START, END); + ByteBuffer second = encoder.encode(write("large.jfr", large), START, END); + + assertNotSame(first.array(), second.array()); + assertPayload(second, large); + } + + @Test + void shrinksBufferAboveRetentionCap() throws IOException { + byte[] large = randomBytes(9 * 1024 * 1024); + byte[] small = randomBytes(1_000); + + ByteBuffer first = encoder.encode(write("large.jfr", large), START, END); + int largeCapacity = first.array().length; + ByteBuffer second = encoder.encode(write("small.jfr", small), START, END); + + assertTrue(second.array().length < largeCapacity); + assertPayload(second, small); + } + + @Test + void writesZeroValueEntryInEveryDictionaryTable() throws IOException { + ByteBuffer encoded = encoder.encode(write("a.jfr", randomBytes(100)), START, END); + byte[] message = new byte[encoded.remaining()]; + encoded.duplicate().get(message); + + Map dictionary = + parse(parse(message).get(OtlpProtoFields.ProfilesData.DICTIONARY)); + int[] tables = { + OtlpProtoFields.ProfilesDictionary.MAPPING_TABLE, + OtlpProtoFields.ProfilesDictionary.LOCATION_TABLE, + OtlpProtoFields.ProfilesDictionary.FUNCTION_TABLE, + OtlpProtoFields.ProfilesDictionary.LINK_TABLE, + OtlpProtoFields.ProfilesDictionary.ATTRIBUTE_TABLE, + OtlpProtoFields.ProfilesDictionary.STACK_TABLE + }; + for (int table : tables) { + assertArrayEquals(new byte[0], dictionary.get(table), "table field " + table); + } + } + + private static void assertPayload(ByteBuffer encoded, byte[] expectedJfr) { + byte[] message = new byte[encoded.remaining()]; + encoded.duplicate().get(message); + + Map profilesData = parse(message); + assertNotNull(profilesData.get(OtlpProtoFields.ProfilesData.DICTIONARY)); + Map resourceProfiles = + parse(profilesData.get(OtlpProtoFields.ProfilesData.RESOURCE_PROFILES)); + String resource = + new String( + resourceProfiles.get(OtlpProtoFields.ResourceProfiles.RESOURCE), + StandardCharsets.ISO_8859_1); + assertTrue(resource.contains("test-service")); + Map scopeProfiles = + parse(resourceProfiles.get(OtlpProtoFields.ResourceProfiles.SCOPE_PROFILES)); + Map profile = parse(scopeProfiles.get(OtlpProtoFields.ScopeProfiles.PROFILES)); + + assertEquals( + "jfr", + new String( + profile.get(OtlpProtoFields.Profile.ORIGINAL_PAYLOAD_FORMAT), StandardCharsets.UTF_8)); + assertEquals(16, profile.get(OtlpProtoFields.Profile.PROFILE_ID).length); + assertArrayEquals(expectedJfr, profile.get(OtlpProtoFields.Profile.ORIGINAL_PAYLOAD)); + } + + /** + * Returns the length-delimited fields of a message by field number; fails if any length prefix + * does not match the bytes that follow, which is what a broken precomputed length would cause. + */ + private static Map parse(byte[] message) { + Map fields = new HashMap<>(); + int[] pos = {0}; + while (pos[0] < message.length) { + long tag = readVarint(message, pos); + int fieldNumber = (int) (tag >>> 3); + switch ((int) (tag & 7)) { + case 0: + readVarint(message, pos); + break; + case 1: + pos[0] += 8; + break; + case 2: + int length = (int) readVarint(message, pos); + fields.put(fieldNumber, Arrays.copyOfRange(message, pos[0], pos[0] + length)); + pos[0] += length; + break; + case 5: + pos[0] += 4; + break; + default: + throw new AssertionError("unexpected wire type in tag " + tag); + } + } + assertEquals(message.length, pos[0], "message length does not match its fields"); + return fields; + } + + private static long readVarint(byte[] buf, int[] pos) { + long value = 0; + for (int shift = 0; ; shift += 7) { + byte b = buf[pos[0]++]; + value |= (long) (b & 0x7F) << shift; + if ((b & 0x80) == 0) { + return value; + } + } + } + + private Path write(String name, byte[] content) throws IOException { + return Files.write(tempDir.resolve(name), content); + } + + private static byte[] randomBytes(int size) { + byte[] bytes = new byte[size]; + new Random(size).nextBytes(bytes); + return bytes; + } +} diff --git a/dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/OtlpProfileUploaderTest.java b/dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/OtlpProfileUploaderTest.java new file mode 100644 index 00000000000..f7d413cdb7e --- /dev/null +++ b/dd-java-agent/agent-profiling/profiling-uploader/src/test/java/com/datadog/profiling/uploader/OtlpProfileUploaderTest.java @@ -0,0 +1,243 @@ +/* + * Copyright 2026 Datadog + * + * Licensed 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 com.datadog.profiling.uploader; + +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_MODE; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.mockito.Mockito.withSettings; + +import datadog.trace.api.Config; +import datadog.trace.api.config.OtlpConfig; +import datadog.trace.api.config.ProfilingConfig; +import datadog.trace.api.profiling.ProfilingSnapshot; +import datadog.trace.api.profiling.RecordingData; +import datadog.trace.api.profiling.RecordingInputStream; +import datadog.trace.api.profiling.RecordingType; +import datadog.trace.bootstrap.config.provider.ConfigProvider; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.io.PrintStream; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.time.Instant; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import okhttp3.mockwebserver.Dispatcher; +import okhttp3.mockwebserver.MockResponse; +import okhttp3.mockwebserver.MockWebServer; +import okhttp3.mockwebserver.RecordedRequest; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.mockito.junit.jupiter.MockitoSettings; +import org.mockito.quality.Strictness; +import org.mockito.stubbing.Answer; + +@ExtendWith(MockitoExtension.class) +@MockitoSettings(strictness = Strictness.LENIENT) +public class OtlpProfileUploaderTest { + + private static final String RECORDING_RESOURCE = "/test-recording.jfr"; + private static final RecordingType RECORDING_TYPE = RecordingType.CONTINUOUS; + private static final String RECORDING_NAME = "test-recording"; + private static final int PROFILE_START = 1000; + private static final int PROFILE_END = 1100; + + private final Duration REQUEST_TIMEOUT = Duration.ofSeconds(10); + private final Duration TERMINATION_TIMEOUT = REQUEST_TIMEOUT.plus(Duration.ofSeconds(5)); + + @Mock private Config config; + @Mock private ConfigProvider configProvider; + + private final MockWebServer server = new MockWebServer(); + private String otlpUrl; + + private OtlpProfileUploader uploader; + + @BeforeEach + public void setup() throws IOException { + server.start(); + otlpUrl = server.url("/v1/profiles").toString(); + + when(config.getFinalProfilingUrl()).thenReturn("http://localhost:8126"); + when(config.getProfilingUploadTimeout()).thenReturn((int) REQUEST_TIMEOUT.getSeconds()); + when(config.getProfilingProxyHost()).thenReturn(null); + when(config.getProfilingProxyPort()).thenReturn(8080); + when(config.getProfilingProxyUsername()).thenReturn(null); + when(config.getProfilingProxyPassword()).thenReturn(null); + + when(config.getServiceName()).thenReturn("test-service"); + when(config.getEnv()).thenReturn(""); + when(config.getVersion()).thenReturn(""); + when(config.isReportHostName()).thenReturn(false); + + when(config.getOtlpProfilesProtocol()).thenReturn(OtlpConfig.Protocol.HTTP_PROTOBUF); + when(config.getOtlpProfilesEndpoint()).thenReturn(otlpUrl); + when(config.getOtlpProfilesHeaders()).thenReturn(Collections.emptyMap()); + when(config.getOtlpProfilesTimeout()).thenReturn((int) REQUEST_TIMEOUT.getSeconds()); + when(config.getOtlpProfilesCompression()).thenReturn(OtlpConfig.Compression.NONE); + + when(configProvider.getBoolean(PROFILING_OTLP_ENABLED, false)).thenReturn(true); + when(configProvider.getEnum( + PROFILING_OTLP_MODE, ProfilingConfig.OtlpMode.class, ProfilingConfig.OtlpMode.LIGHT)) + .thenReturn(ProfilingConfig.OtlpMode.LIGHT); + + uploader = + new OtlpProfileUploader(config, configProvider, (int) TERMINATION_TIMEOUT.getSeconds()); + } + + @AfterEach + public void teardown() throws IOException { + uploader.shutdown(); + server.shutdown(); + } + + @Test + public void testDisabledUploader() throws Exception { + when(configProvider.getBoolean(PROFILING_OTLP_ENABLED, false)).thenReturn(false); + + OtlpProfileUploader disabledUploader = + new OtlpProfileUploader(config, configProvider, (int) TERMINATION_TIMEOUT.getSeconds()); + + RecordingData data = mockRecordingData(); + + disabledUploader.onNewData(RECORDING_TYPE, data, true); + + assertEquals(0, server.getRequestCount()); + verify(data).release(); + + disabledUploader.shutdown(); + } + + @Test + public void testLightweightUploadCarriesResourceAttributes() throws Exception { + // OtlpHttpSender retries up to 5 times (HttpRetryPolicy.Factory(5, ...)). Serve an endless + // stream of 200s so a retried attempt never blocks on an empty response queue (which would + // otherwise stall 10s per retry and inflate getRequestCount past 1 under CI contention). + server.setDispatcher( + new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest request) { + return new MockResponse(); + } + }); + + RecordingData data = mockRecordingData(); + + uploader.onNewData(RECORDING_TYPE, data, true); + + // The uploader initiates exactly one upload; getRequestCount() may exceed 1 only because of + // transport-level retries, which are outside the scope of this test. + assertTrue(server.getRequestCount() >= 1, "expected at least one upload request"); + RecordedRequest request = server.takeRequest(REQUEST_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS); + assertNotNull(request); + byte[] body = request.getBody().readByteArray(); + // String fields are encoded inline on the wire, so the attribute key/value pairs are + // searchable in the raw payload + assertTrue(new String(body, StandardCharsets.ISO_8859_1).contains("service.name")); + assertTrue(new String(body, StandardCharsets.ISO_8859_1).contains("test-service")); + assertTrue(new String(body, StandardCharsets.ISO_8859_1).contains("telemetry.sdk.name")); + verify(data).release(); + } + + @Test + public void syncUploadSkipsWhenAsyncExportDoesNotFinishInTime() throws Exception { + CountDownLatch requestReceived = new CountDownLatch(1); + CountDownLatch unblock = new CountDownLatch(1); + server.setDispatcher( + new Dispatcher() { + @Override + public MockResponse dispatch(RecordedRequest request) throws InterruptedException { + requestReceived.countDown(); + unblock.await(); + return new MockResponse(); + } + }); + OtlpProfileUploader shortWaitUploader = new OtlpProfileUploader(config, configProvider, 1); + try { + shortWaitUploader.onNewData(RECORDING_TYPE, mockRecordingData(), false); + assertTrue(requestReceived.await(REQUEST_TIMEOUT.toMillis(), TimeUnit.MILLISECONDS)); + + RecordingData syncData = mockRecordingData(); + shortWaitUploader.onNewData(RECORDING_TYPE, syncData, true); + + verify(syncData, never()).getStream(); + verify(syncData).release(); + } finally { + unblock.countDown(); + shortWaitUploader.shutdown(); + } + } + + @Test + public void warnsThatOtlpProfilesAreNotProductionReadyWhenEnabled() { + String log = captureStderr(() -> new OtlpProfileUploader(config, configProvider, 1).shutdown()); + + assertTrue(log.contains("WARN"), log); + assertTrue(log.contains("OTLP profiles export is enabled"), log); + assertTrue(log.contains("do not use it in production"), log); + } + + @Test + public void doesNotWarnWhenDisabled() { + when(configProvider.getBoolean(PROFILING_OTLP_ENABLED, false)).thenReturn(false); + + String log = captureStderr(() -> new OtlpProfileUploader(config, configProvider, 1).shutdown()); + + assertFalse(log.contains("OTLP profiles export is enabled"), log); + } + + // the test classpath binds slf4j-simple, which writes to the current System.err + private static String captureStderr(Runnable action) { + PrintStream original = System.err; + ByteArrayOutputStream captured = new ByteArrayOutputStream(); + System.setErr(new PrintStream(captured, true)); + try { + action.run(); + } finally { + System.setErr(original); + } + return new String(captured.toByteArray(), StandardCharsets.UTF_8); + } + + private RecordingData mockRecordingData() throws IOException { + final RecordingData recordingData = mock(RecordingData.class, withSettings().lenient()); + when(recordingData.getStream()) + .then( + (Answer) + invocation -> + new RecordingInputStream(getClass().getResourceAsStream(RECORDING_RESOURCE))); + when(recordingData.getName()).thenReturn(RECORDING_NAME); + when(recordingData.getStart()).thenReturn(Instant.ofEpochSecond(PROFILE_START)); + when(recordingData.getEnd()).thenReturn(Instant.ofEpochSecond(PROFILE_END)); + when(recordingData.getKind()).thenReturn(ProfilingSnapshot.Kind.PERIODIC); + when(recordingData.getPath()).thenReturn(null); // Force stream-based conversion + return recordingData; + } +} diff --git a/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java b/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java index 9175e0bc40f..5cbcdfe8865 100644 --- a/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java +++ b/dd-java-agent/agent-profiling/src/main/java/com/datadog/profiling/agent/ProfilingAgent.java @@ -3,6 +3,8 @@ import static datadog.environment.JavaVirtualMachine.isJavaVersion; import static datadog.environment.JavaVirtualMachine.isJavaVersionAtLeast; import static datadog.environment.JavaVirtualMachine.isOracleJDK8; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED_DEFAULT; import static datadog.trace.api.config.ProfilingConfig.PROFILING_SCRUB_ENABLED; import static datadog.trace.api.config.ProfilingConfig.PROFILING_SCRUB_ENABLED_DEFAULT; import static datadog.trace.api.config.ProfilingConfig.PROFILING_SCRUB_FAIL_OPEN; @@ -19,6 +21,7 @@ import com.datadog.profiling.controller.ProfilingSystem; import com.datadog.profiling.controller.UnsupportedEnvironmentException; import com.datadog.profiling.controller.jfr.JFRAccess; +import com.datadog.profiling.uploader.OtlpProfileUploader; import com.datadog.profiling.uploader.ProfileUploader; import com.datadog.profiling.utils.Timestamper; import datadog.trace.api.Config; @@ -55,6 +58,7 @@ public class ProfilingAgent { private static volatile ProfilingSystem profiler; private static volatile ProfileUploader uploader; + private static volatile OtlpProfileUploader otlpUploader; private static class DataDumper implements RecordingDataListener { private final Path path; @@ -144,10 +148,15 @@ public static synchronized boolean run(final boolean earlyStart, Instrumentation final Controller controller = CompositeController.build(configProvider, context); String dumpPath = configProvider.getString(ProfilingConfig.PROFILING_DEBUG_DUMP_PATH); + // local: only the listener chain below consumes the dumper DataDumper dumper = dumpPath != null ? new DataDumper(Paths.get(dumpPath)) : null; uploader = new ProfileUploader(config, configProvider); + if (configProvider.getBoolean(PROFILING_OTLP_ENABLED, PROFILING_OTLP_ENABLED_DEFAULT)) { + otlpUploader = new OtlpProfileUploader(config, configProvider); + } + RecordingDataListener listener = uploader::upload; if (dumper != null) { RecordingDataListener upload = listener; @@ -157,7 +166,13 @@ public static synchronized boolean run(final boolean earlyStart, Instrumentation upload.onNewData(type, data, sync); }; } - // Scrubber wraps the combined dumper+uploader so debug dumps also contain scrubbed data + if (otlpUploader != null) { + OtlpProfileUploader otlp = otlpUploader; + listener = + withOtlpUpload(listener, (type, data, sync) -> otlp.upload(type, data, sync, null)); + } + // the scrubber must wrap the OTLP listener so OTLP receives the scrubbed copy, not the + // raw recording; it also wraps the combined dumper+uploader so debug dumps stay scrubbed // Oracle JDK 8 JFR format has quirks that make scrubbing unreliable — skip it to avoid // corrupting customer data if (configProvider.getBoolean(PROFILING_SCRUB_ENABLED, PROFILING_SCRUB_ENABLED_DEFAULT) @@ -197,7 +212,7 @@ public static synchronized boolean run(final boolean earlyStart, Instrumentation This means that if/when we implement functionality to manually shutdown profiler we would need to not forget to add code that removes this shutdown hook from JVM. */ - Runtime.getRuntime().addShutdownHook(new ShutdownHook(profiler, uploader)); + Runtime.getRuntime().addShutdownHook(new ShutdownHook(profiler, uploader, otlpUploader)); } catch (final IllegalStateException ex) { // The JVM is already shutting down. } @@ -209,6 +224,32 @@ public static synchronized boolean run(final boolean earlyStart, Instrumentation return false; } + /** + * Runs {@code downstream} first and then {@code otlp}, so OTLP export never delays the classic + * upload. The OTLP reference is taken before downstream runs, since downstream may release the + * base reference and free the recording; {@code otlp} must release that reference. If the + * reference cannot be taken, only {@code downstream} runs. + */ + static RecordingDataListener withOtlpUpload( + RecordingDataListener downstream, RecordingDataListener otlp) { + return (type, data, sync) -> { + boolean retained = false; + try { + data.retain(); + retained = true; + } catch (IllegalStateException e) { + log.warn(SEND_TELEMETRY, "OTLP upload skipped, JFR upload will continue", e); + } + try { + downstream.onNewData(type, data, sync); + } finally { + if (retained) { + otlp.onNewData(type, data, sync); + } + } + }; + } + private static RecordingDataListener wrapWithScrubber( RecordingDataListener listener, List excludeEventTypes, boolean failOpen) { try { @@ -226,17 +267,20 @@ private static boolean isStartForceFirstSafe() { } public static void shutdown() { - shutdown(profiler, uploader, false); + shutdown(profiler, uploader, otlpUploader, false); } public static void shutdown(boolean snapshot) { - shutdown(profiler, uploader, snapshot); + shutdown(profiler, uploader, otlpUploader, snapshot); } private static final AtomicBoolean shutDownFlag = new AtomicBoolean(); private static void shutdown( - ProfilingSystem profiler, ProfileUploader uploader, boolean snapshot) { + ProfilingSystem profiler, + ProfileUploader uploader, + OtlpProfileUploader otlpUploader, + boolean snapshot) { if (shutDownFlag.compareAndSet(false, true)) { if (profiler != null) { profiler.shutdown(snapshot); @@ -245,6 +289,10 @@ private static void shutdown( if (uploader != null) { uploader.shutdown(); } + + if (otlpUploader != null) { + otlpUploader.shutdown(); + } } } @@ -252,16 +300,21 @@ private static class ShutdownHook extends Thread { private final WeakReference profilerRef; private final WeakReference uploaderRef; + private final WeakReference otlpUploaderRef; - private ShutdownHook(final ProfilingSystem profiler, final ProfileUploader uploader) { + private ShutdownHook( + final ProfilingSystem profiler, + final ProfileUploader uploader, + final OtlpProfileUploader otlpUploader) { super(AGENT_THREAD_GROUP, "dd-profiler-shutdown-hook"); profilerRef = new WeakReference<>(profiler); uploaderRef = new WeakReference<>(uploader); + otlpUploaderRef = new WeakReference<>(otlpUploader); } @Override public void run() { - shutdown(profilerRef.get(), uploaderRef.get(), false); + shutdown(profilerRef.get(), uploaderRef.get(), otlpUploaderRef.get(), false); } } } diff --git a/dd-java-agent/agent-profiling/src/test/java/com/datadog/profiling/agent/OtlpUploadListenerTest.java b/dd-java-agent/agent-profiling/src/test/java/com/datadog/profiling/agent/OtlpUploadListenerTest.java new file mode 100644 index 00000000000..adf685f0731 --- /dev/null +++ b/dd-java-agent/agent-profiling/src/test/java/com/datadog/profiling/agent/OtlpUploadListenerTest.java @@ -0,0 +1,99 @@ +package com.datadog.profiling.agent; + +import static com.datadog.profiling.agent.ProfilingAgent.withOtlpUpload; +import static java.util.Arrays.asList; +import static java.util.Collections.singletonList; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import datadog.trace.api.profiling.ProfilingSnapshot; +import datadog.trace.api.profiling.RecordingData; +import datadog.trace.api.profiling.RecordingDataListener; +import datadog.trace.api.profiling.RecordingInputStream; +import datadog.trace.api.profiling.RecordingType; +import java.io.ByteArrayInputStream; +import java.time.Instant; +import java.util.ArrayList; +import java.util.List; +import org.junit.jupiter.api.Test; + +class OtlpUploadListenerTest { + + private final List calls = new ArrayList<>(); + private final TestRecordingData data = new TestRecordingData(); + + @Test + void classicUploadRunsBeforeOtlpAndRecordingOutlivesIt() { + RecordingDataListener downstream = + (type, d, sync) -> { + calls.add("downstream"); + d.release(); // drops the base reference, as the classic uploader does + }; + RecordingDataListener otlp = + (type, d, sync) -> { + calls.add("otlp:released=" + data.released); + d.release(); + }; + + withOtlpUpload(downstream, otlp).onNewData(RecordingType.CONTINUOUS, data, false); + + assertEquals(asList("downstream", "otlp:released=false"), calls); + assertTrue(data.released); + } + + @Test + void otlpRunsWhenClassicUploadThrows() { + RecordingDataListener downstream = + (type, d, sync) -> { + d.release(); + throw new IllegalArgumentException("boom"); + }; + RecordingDataListener otlp = + (type, d, sync) -> { + calls.add("otlp"); + d.release(); + }; + + assertThrows( + IllegalArgumentException.class, + () -> withOtlpUpload(downstream, otlp).onNewData(RecordingType.CONTINUOUS, data, false)); + + assertEquals(singletonList("otlp"), calls); + assertTrue(data.released); + } + + @Test + void skipsOtlpWhenRecordingAlreadyReleased() { + data.release(); + RecordingDataListener downstream = (type, d, sync) -> calls.add("downstream"); + RecordingDataListener otlp = (type, d, sync) -> calls.add("otlp"); + + withOtlpUpload(downstream, otlp).onNewData(RecordingType.CONTINUOUS, data, false); + + assertEquals(singletonList("downstream"), calls); + } + + private static final class TestRecordingData extends RecordingData { + boolean released; + + TestRecordingData() { + super(Instant.EPOCH, Instant.EPOCH, ProfilingSnapshot.Kind.PERIODIC); + } + + @Override + public RecordingInputStream getStream() { + return new RecordingInputStream(new ByteArrayInputStream(new byte[0])); + } + + @Override + protected void doRelease() { + released = true; + } + + @Override + public String getName() { + return "test"; + } + } +} diff --git a/dd-trace-api/build.gradle.kts b/dd-trace-api/build.gradle.kts index faf02823ec9..8896010ba12 100644 --- a/dd-trace-api/build.gradle.kts +++ b/dd-trace-api/build.gradle.kts @@ -41,6 +41,7 @@ extra["excludedClassesCoverage"] = listOf( "datadog.trace.api.civisibility.noop.NoOpDDTestSuite", "datadog.trace.api.config.AIGuardConfig", "datadog.trace.api.config.ProfilingConfig", + "datadog.trace.api.config.ProfilingConfig.OtlpMode", "datadog.trace.api.interceptor.MutableSpan", "datadog.trace.api.profiling.Profiling", "datadog.trace.api.profiling.Profiling.NoOp", diff --git a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java index bf5bab30d05..12bf2b9bb72 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/ConfigDefaults.java @@ -136,6 +136,13 @@ public final class ConfigDefaults { public static final String DEFAULT_OTLP_HTTP_PORT = "4318"; public static final String DEFAULT_OTLP_GRPC_PORT = "4317"; + // Provisional — the OTLP profiles proto is still in v1development; update when stabilized. + // Current collectors expose /v1development/profiles and the v1development ProfilesService. + public static final String DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT = "v1development/profiles"; + public static final String DEFAULT_OTLP_GRPC_PROFILES_ENDPOINT = + "opentelemetry.proto.collector.profiles.v1development.ProfilesService/Export"; + public static final int DEFAULT_OTLP_PROFILES_TIMEOUT = 30_000; // ms + static final int DEFAULT_DOGSTATSD_START_DELAY = 15; // seconds static final boolean DEFAULT_HEALTH_METRICS_ENABLED = true; diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/OtlpConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/OtlpConfig.java index 90fd9fd046f..3abf06f9ea2 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/config/OtlpConfig.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/config/OtlpConfig.java @@ -42,6 +42,13 @@ public final class OtlpConfig { public static final String OTLP_TRACES_COMPRESSION = "otlp.traces.compression"; public static final String OTLP_TRACES_TIMEOUT = "otlp.traces.timeout"; + // OTLP Profiles — provisional, proto is still v1development + public static final String OTLP_PROFILES_ENDPOINT = "otlp.profiles.endpoint"; + public static final String OTLP_PROFILES_HEADERS = "otlp.profiles.headers"; + public static final String OTLP_PROFILES_PROTOCOL = "otlp.profiles.protocol"; + public static final String OTLP_PROFILES_COMPRESSION = "otlp.profiles.compression"; + public static final String OTLP_PROFILES_TIMEOUT = "otlp.profiles.timeout"; + public enum Protocol { GRPC, HTTP_PROTOBUF, diff --git a/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java b/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java index 37dc6c3c500..2d479c14320 100644 --- a/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java +++ b/dd-trace-api/src/main/java/datadog/trace/api/config/ProfilingConfig.java @@ -287,5 +287,22 @@ public final class ProfilingConfig { public static final String PROFILING_SCRUB_EXCLUDE_EVENTS = "profiling.scrub.exclude-events"; + // OTLP Profiles Format Support + public static final String PROFILING_OTLP_ENABLED = "profiling.otlp.enabled"; + public static final boolean PROFILING_OTLP_ENABLED_DEFAULT = false; + + // LIGHT = raw JFR as the original_payload blob, no conversion — the profile carries no samples, + // so only a backend that parses the embedded JFR can use it; + // FULL = converted samples + raw JFR blob; CONVERTED = converted samples only + public static final String PROFILING_OTLP_MODE = "profiling.otlp.mode"; + + public enum OtlpMode { + LIGHT, + FULL, + CONVERTED + } + + public static final ProfilingConfig.OtlpMode PROFILING_OTLP_MODE_DEFAULT = OtlpMode.LIGHT; + private ProfilingConfig() {} } diff --git a/internal-api/src/main/java/datadog/trace/api/Config.java b/internal-api/src/main/java/datadog/trace/api/Config.java index 3471750c347..7c0aa16a09b 100644 --- a/internal-api/src/main/java/datadog/trace/api/Config.java +++ b/internal-api/src/main/java/datadog/trace/api/Config.java @@ -133,7 +133,9 @@ import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_LOGS_ENDPOINT; import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_METRICS_ENDPOINT; import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_PORT; +import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT; import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_TRACES_ENDPOINT; +import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_PROFILES_TIMEOUT; import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_TRACES_TIMEOUT; import static datadog.trace.api.ConfigDefaults.DEFAULT_PARTIAL_FLUSH_MIN_SPANS; import static datadog.trace.api.ConfigDefaults.DEFAULT_PERF_METRICS_ENABLED; @@ -508,6 +510,11 @@ import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_PROTOCOL; import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_TEMPORALITY_PREFERENCE; import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_TIMEOUT; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_COMPRESSION; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_ENDPOINT; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_HEADERS; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_PROTOCOL; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_TIMEOUT; import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_COMPRESSION; import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_ENDPOINT; import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_HEADERS; @@ -1051,6 +1058,12 @@ public static String getHostName() { private final OtlpConfig.Compression otlpTracesCompression; private final int otlpTracesTimeout; + private final String otlpProfilesEndpoint; + private final Map otlpProfilesHeaders; + private final OtlpConfig.Protocol otlpProfilesProtocol; + private final OtlpConfig.Compression otlpProfilesCompression; + private final int otlpProfilesTimeout; + // These values are default-ed to those of jmx fetch values as needed private final boolean healthMetricsEnabled; private final String healthMetricsStatsdHost; @@ -2254,6 +2267,40 @@ && isMetricsOtelEnabled() } otlpTracesEndpoint = otlpTracesEndpointFromEnvironment; + // OTLP Profiles config — mirrors the traces/logs/metrics pattern + int profilesTimeout = + configProvider.getInteger(OTLP_PROFILES_TIMEOUT, DEFAULT_OTLP_PROFILES_TIMEOUT); + if (profilesTimeout < 0) { + log.warn( + "Invalid OTLP profiles timeout: {}. The value must not be negative", profilesTimeout); + profilesTimeout = DEFAULT_OTLP_PROFILES_TIMEOUT; + } + otlpProfilesTimeout = profilesTimeout; + + otlpProfilesHeaders = configProvider.getMergedMap(OTLP_PROFILES_HEADERS, '='); + otlpProfilesProtocol = + configProvider.getEnum( + OTLP_PROFILES_PROTOCOL, OtlpConfig.Protocol.class, OtlpConfig.Protocol.HTTP_PROTOBUF); + otlpProfilesCompression = + configProvider.getEnum( + OTLP_PROFILES_COMPRESSION, OtlpConfig.Compression.class, OtlpConfig.Compression.NONE); + + String otlpProfilesEndpointFromEnvironment = configProvider.getString(OTLP_PROFILES_ENDPOINT); + if (otlpProfilesEndpointFromEnvironment == null) { + if (otlpProfilesProtocol == OtlpConfig.Protocol.GRPC) { + otlpProfilesEndpointFromEnvironment = "http://" + agentHost + ':' + DEFAULT_OTLP_GRPC_PORT; + } else { + otlpProfilesEndpointFromEnvironment = + "http://" + + agentHost + + ':' + + DEFAULT_OTLP_HTTP_PORT + + '/' + + DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT; + } + } + otlpProfilesEndpoint = otlpProfilesEndpointFromEnvironment; + // Runtime metrics are disabled if Otel metrics are enabled and the metrics exporter is none runtimeMetricsEnabled = configProvider.getBoolean(RUNTIME_METRICS_ENABLED, true); @@ -5998,6 +6045,26 @@ public int getOtlpTracesTimeout() { return otlpTracesTimeout; } + public String getOtlpProfilesEndpoint() { + return otlpProfilesEndpoint; + } + + public Map getOtlpProfilesHeaders() { + return otlpProfilesHeaders; + } + + public OtlpConfig.Protocol getOtlpProfilesProtocol() { + return otlpProfilesProtocol; + } + + public OtlpConfig.Compression getOtlpProfilesCompression() { + return otlpProfilesCompression; + } + + public int getOtlpProfilesTimeout() { + return otlpProfilesTimeout; + } + public boolean isRuleEnabled(final String name) { return isRuleEnabled(name, true); } diff --git a/internal-api/src/main/java/datadog/trace/api/telemetry/OtlpTelemetry.java b/internal-api/src/main/java/datadog/trace/api/telemetry/OtlpTelemetry.java index a36bcff67c8..3323a10a4d0 100644 --- a/internal-api/src/main/java/datadog/trace/api/telemetry/OtlpTelemetry.java +++ b/internal-api/src/main/java/datadog/trace/api/telemetry/OtlpTelemetry.java @@ -8,11 +8,15 @@ import java.util.List; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.atomic.LongAccumulator; import java.util.concurrent.atomic.LongAdder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** Collects telemetry metrics for the OTLP trace, metrics, and log exporters. */ public class OtlpTelemetry implements MetricCollector { private static final String NAMESPACE = "tracers"; + private static final Logger log = LoggerFactory.getLogger(OtlpTelemetry.class); private static final OtlpTelemetry INSTANCE = new OtlpTelemetry(); @@ -24,6 +28,16 @@ public static OtlpTelemetry getInstance() { private final String[] metricsTags = tagsFor(Config.get().getOtlpMetricsProtocol()); private final String[] logsTags = tagsFor(Config.get().getOtlpLogsProtocol()); + private final String[] profilesTags = tagsFor(Config.get().getOtlpProfilesProtocol()); + private final ExportCounters profilesExport = new ExportCounters("profiles"); + + // JFR→OTLP conversion timings, reported as avg/max gauges per flush window + private final LongAdder profilesConversionNanosTotal = new LongAdder(); + private final LongAdder profilesConversionCount = new LongAdder(); + // guards the (nanosTotal, count) pair so window snapshots are consistent + private final Object conversionMetricsLock = new Object(); + private final LongAccumulator profilesConversionNanosMax = new LongAccumulator(Long::max, 0); + private final ExportCounters tracesExport = new ExportCounters("traces"); private final ExportCounters metricsExport = new ExportCounters("metrics"); private final LongAdder logRecords = new LongAdder(); @@ -44,10 +58,29 @@ public void onMetricsExportAttempt() { metricsExport.attempts.increment(); } + public void onProfilesExportAttempt() { + profilesExport.attempts.increment(); + } + public void onMetricsExportComplete(boolean success) { metricsExport.complete(success); } + public void onProfilesExportComplete(boolean success) { + profilesExport.complete(success); + } + + /** Records the duration of a single JFR→OTLP profile conversion. */ + public void onProfilesConversion(long nanos) { + // (total, count, max) must stay a consistent window snapshot; the conversion metrics lock + // also guards the snapshot+reset in prepareMetrics() + synchronized (conversionMetricsLock) { + profilesConversionNanosTotal.add(nanos); + profilesConversionCount.increment(); + profilesConversionNanosMax.accumulate(nanos); + } + } + public void onLogRecordsSubmitted(long count) { if (count > 0) { logRecords.add(count); @@ -64,12 +97,43 @@ private static String[] tagsFor(OtlpConfig.Protocol protocol) { public void prepareMetrics() { tracesExport.stageInto(telemetryQueue, tracesTags); metricsExport.stageInto(telemetryQueue, metricsTags); + profilesExport.stageInto(telemetryQueue, profilesTags); + long conversionCount; + long conversionNanosTotal; + long maxNanos; + synchronized (conversionMetricsLock) { + // snapshot+reset (count, total, max) atomically so each flush window reports exactly the + // conversions that completed within it + conversionCount = profilesConversionCount.sumThenReset(); + conversionNanosTotal = profilesConversionNanosTotal.sumThenReset(); + maxNanos = profilesConversionNanosMax.getThenReset(); + } + if (conversionCount > 0) { + double avgNanos = conversionNanosTotal / (double) conversionCount; + offerMetric( + new OtlpMetric( + "otel.profiles_conversion_ms", "gauge", nanosToMillis(avgNanos), profilesTags)); + offerMetric( + new OtlpMetric( + "otel.profiles_conversion_max_ms", "gauge", nanosToMillis(maxNanos), profilesTags)); + } long logRecordCount = logRecords.sumThenReset(); if (logRecordCount > 0) { - telemetryQueue.offer(new OtlpMetric("otel.log_records", logRecordCount, logsTags)); + offerMetric(new OtlpMetric("otel.log_records", logRecordCount, logsTags)); + } + } + + // the queue is bounded and shared by all OTLP surfaces; make a rejected offer visible + private void offerMetric(OtlpMetric metric) { + if (!telemetryQueue.offer(metric)) { + log.debug("Telemetry queue full, dropping metric {}", metric.metricName); } } + private static double nanosToMillis(double nanos) { + return nanos / 1_000_000.0; + } + @Override public Collection drain() { if (telemetryQueue.isEmpty()) { @@ -109,15 +173,19 @@ void stageInto(BlockingQueue out, String[] tags) { private static void addIfNonZero( BlockingQueue out, String metricName, LongAdder counter, String[] tags) { long value = counter.sumThenReset(); - if (value > 0) { - out.offer(new OtlpMetric(metricName, value, tags)); + if (value > 0 && !out.offer(new OtlpMetric(metricName, value, tags))) { + log.debug("Telemetry queue full, dropping metric {}", metricName); } } } public static class OtlpMetric extends MetricCollector.Metric { public OtlpMetric(String metricName, long value, String... tags) { - super(NAMESPACE, true, metricName, "count", value, tags); + this(metricName, "count", value, tags); + } + + OtlpMetric(String metricName, String type, Number value, String... tags) { + super(NAMESPACE, true, metricName, type, value, tags); } } } diff --git a/internal-api/src/test/java/datadog/trace/api/telemetry/OtlpTelemetryTest.java b/internal-api/src/test/java/datadog/trace/api/telemetry/OtlpTelemetryTest.java index 53877bd8dc6..18888b70cdb 100644 --- a/internal-api/src/test/java/datadog/trace/api/telemetry/OtlpTelemetryTest.java +++ b/internal-api/src/test/java/datadog/trace/api/telemetry/OtlpTelemetryTest.java @@ -1,6 +1,7 @@ package datadog.trace.api.telemetry; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertTrue; import java.util.Collection; @@ -86,6 +87,46 @@ void metricsExportMetricsUseExpectedNames() { assertEquals(1L, valuesByName.get("otel.metrics_export_failures")); } + @Test + void onProfilesConversionQueuesAvgAndMaxGauges() { + collector.onProfilesConversion(50_000_000L); // 50 ms + collector.onProfilesConversion(150_000_000L); // 150 ms + collector.prepareMetrics(); + + Collection metrics = collector.drain(); + + Map gaugesByName = new HashMap<>(); + for (OtlpTelemetry.OtlpMetric metric : metrics) { + gaugesByName.put(metric.metricName, metric); + } + assertEquals(2, gaugesByName.size()); + + OtlpTelemetry.OtlpMetric avg = gaugesByName.get("otel.profiles_conversion_ms"); + assertEquals("gauge", avg.type); + assertEquals(100.0, avg.value.doubleValue(), 0.001); + assertTrue(avg.tags.contains("protocol:http")); + + OtlpTelemetry.OtlpMetric max = gaugesByName.get("otel.profiles_conversion_max_ms"); + assertEquals("gauge", max.type); + assertEquals(150.0, max.value.doubleValue(), 0.001); + assertTrue(max.tags.contains("protocol:http")); + } + + @Test + void onProfilesConversionOmittedWhenNoConversions() { + // stage a different metric so a non-empty drain proves the gauges were specifically + // omitted rather than nothing being emitted at all + collector.onTracesExportAttempt(); + collector.prepareMetrics(); + + Collection metrics = collector.drain(); + + assertFalse(metrics.isEmpty(), "expected the traces metric to be staged"); + assertTrue( + metrics.stream().noneMatch(m -> m.metricName.startsWith("otel.profiles_conversion")), + "no profiles conversion metrics expected without conversions"); + } + @Test void onLogRecordsSubmittedQueuesCountWithGivenValue() { collector.onLogRecordsSubmitted(5); diff --git a/internal-api/src/test/java/datadog/trace/bootstrap/config/provider/OtelProfilesEnvironmentConfigSourceTest.java b/internal-api/src/test/java/datadog/trace/bootstrap/config/provider/OtelProfilesEnvironmentConfigSourceTest.java new file mode 100644 index 00000000000..93afff7978d --- /dev/null +++ b/internal-api/src/test/java/datadog/trace/bootstrap/config/provider/OtelProfilesEnvironmentConfigSourceTest.java @@ -0,0 +1,116 @@ +package datadog.trace.bootstrap.config.provider; + +import static datadog.trace.api.config.GeneralConfig.RUNTIME_METRICS_ENABLED; +import static datadog.trace.api.config.GeneralConfig.SERVICE_NAME; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_COMPRESSION; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_ENDPOINT; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_HEADERS; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_PROTOCOL; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_TIMEOUT; +import static datadog.trace.test.junit.utils.config.WithConfigExtension.injectEnvConfig; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import datadog.trace.test.junit.utils.config.WithConfigExtension; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(WithConfigExtension.class) +class OtelProfilesEnvironmentConfigSourceTest { + + @Test + void profilesKeysAreNotMappedWhenOtlpProfilingIsDisabled() { + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_ENDPOINT", "http://collector:4318/p", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertNull(source.get(OTLP_PROFILES_ENDPOINT)); + } + + @Test + void signalSpecificProfilesKeysAreMapped() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_ENDPOINT", "http://collector:4318/p", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_HEADERS", "api-key=secret", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_PROTOCOL", "http/protobuf", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_COMPRESSION", "gzip", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_TIMEOUT", "5000", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertEquals("http://collector:4318/p", source.get(OTLP_PROFILES_ENDPOINT)); + assertEquals("api-key=secret", source.get(OTLP_PROFILES_HEADERS)); + assertEquals("http/protobuf", source.get(OTLP_PROFILES_PROTOCOL)); + assertEquals("gzip", source.get(OTLP_PROFILES_COMPRESSION)); + assertEquals("5000", source.get(OTLP_PROFILES_TIMEOUT)); + } + + @Test + void generalEndpointGetsProfilesPathAppended() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4318", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_HEADERS", "api-key=secret", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertEquals( + "http://collector:4318/v1development/profiles", source.get(OTLP_PROFILES_ENDPOINT)); + assertEquals("api-key=secret", source.get(OTLP_PROFILES_HEADERS)); + } + + @Test + void generalEndpointIsKeptAsIsForGrpc() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROTOCOL", "grpc", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4317", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertEquals("http://collector:4317", source.get(OTLP_PROFILES_ENDPOINT)); + } + + @Test + void generalEndpointIsKeptAsIsForDatadogGrpcProtocol() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("DD_OTLP_PROFILES_PROTOCOL", "grpc", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_ENDPOINT", "http://collector:4317", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertEquals("http://collector:4317", source.get(OTLP_PROFILES_ENDPOINT)); + } + + @Test + void otlpProfilingDoesNotMapGeneralOtelEnvironment() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("OTEL_SERVICE_NAME", "otel-service", false); + injectEnvConfig("OTEL_METRICS_EXPORTER", "none", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertNull(source.get(SERVICE_NAME)); + assertNull(source.get(RUNTIME_METRICS_ENABLED)); + } + + @Test + void profilesKeysAreMappedWhenOtelSdkIsDisabled() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("OTEL_SDK_DISABLED", "true", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_ENDPOINT", "http://collector:4318/p", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertEquals("http://collector:4318/p", source.get(OTLP_PROFILES_ENDPOINT)); + } + + @Test + void datadogSettingWinsOverOtelSetting() { + injectEnvConfig("DD_PROFILING_OTLP_ENABLED", "true", false); + injectEnvConfig("DD_OTLP_PROFILES_ENDPOINT", "http://dd:4318/p", false); + injectEnvConfig("OTEL_EXPORTER_OTLP_PROFILES_ENDPOINT", "http://collector:4318/p", false); + + OtelEnvironmentConfigSource source = new OtelEnvironmentConfigSource(); + + assertNull(source.get(OTLP_PROFILES_ENDPOINT)); + } +} diff --git a/metadata/supported-configurations.json b/metadata/supported-configurations.json index 45e12e44c85..10e7e8589f6 100644 --- a/metadata/supported-configurations.json +++ b/metadata/supported-configurations.json @@ -12134,6 +12134,47 @@ "aliases": [] } ], + "OTEL_EXPORTER_OTLP_PROFILES_ENDPOINT": [ + { + "version": "A", + "type": "string", + "default": null, + "aliases": [] + } + ], + "OTEL_EXPORTER_OTLP_PROFILES_HEADERS": [ + { + "version": "A", + "type": "string", + "default": null, + "aliases": [], + "sensitive": true + } + ], + "OTEL_EXPORTER_OTLP_PROFILES_PROTOCOL": [ + { + "version": "A", + "type": "string", + "default": "http/protobuf", + "aliases": [] + } + ], + "OTEL_EXPORTER_OTLP_PROFILES_COMPRESSION": [ + { + "version": "B", + "type": "string", + "default": "none", + "aliases": [] + } + ], + "OTEL_EXPORTER_OTLP_PROFILES_TIMEOUT": [ + { + "version": "A", + "type": "int", + "default": "30000", + "aliases": [] + } + ], "OTEL_EXPORTER_OTLP_METRICS_ENDPOINT": [ { "version": "A", @@ -12255,6 +12296,63 @@ "default": null, "aliases": [] } + ], + "DD_OTLP_PROFILES_ENDPOINT": [ + { + "version": "A", + "type": "string", + "default": null, + "aliases": [] + } + ], + "DD_OTLP_PROFILES_HEADERS": [ + { + "version": "A", + "type": "map", + "default": null, + "aliases": [], + "sensitive": true + } + ], + "DD_OTLP_PROFILES_PROTOCOL": [ + { + "version": "A", + "type": "string", + "default": "HTTP_PROTOBUF", + "aliases": [] + } + ], + "DD_OTLP_PROFILES_COMPRESSION": [ + { + "version": "A", + "type": "string", + "default": "NONE", + "aliases": [] + } + ], + "DD_OTLP_PROFILES_TIMEOUT": [ + { + "version": "A", + "type": "int", + "default": "30000", + "aliases": [] + } + ], + "DD_PROFILING_OTLP_ENABLED": [ + { + "version": "A", + "type": "boolean", + "default": "false", + "aliases": [] + } + ], + "DD_PROFILING_OTLP_MODE": [ + { + "version": "B", + "type": "string", + "default": "LIGHT", + "aliases": [] + } ] }, "deprecations": { diff --git a/utils/config-utils/src/main/java/datadog/trace/bootstrap/config/provider/OtelEnvironmentConfigSource.java b/utils/config-utils/src/main/java/datadog/trace/bootstrap/config/provider/OtelEnvironmentConfigSource.java index 3c7aa53d238..ddbe47664dd 100644 --- a/utils/config-utils/src/main/java/datadog/trace/bootstrap/config/provider/OtelEnvironmentConfigSource.java +++ b/utils/config-utils/src/main/java/datadog/trace/bootstrap/config/provider/OtelEnvironmentConfigSource.java @@ -2,6 +2,7 @@ import static datadog.trace.api.ConfigDefaults.DEFAULT_LOGS_OTEL_ENABLED; import static datadog.trace.api.ConfigDefaults.DEFAULT_METRICS_OTEL_ENABLED; +import static datadog.trace.api.ConfigDefaults.DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT; import static datadog.trace.api.ConfigDefaults.DEFAULT_TRACE_OTEL_ENABLED; import static datadog.trace.api.config.GeneralConfig.ENV; import static datadog.trace.api.config.GeneralConfig.LOG_LEVEL; @@ -33,6 +34,11 @@ import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_PROTOCOL; import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_TEMPORALITY_PREFERENCE; import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_TIMEOUT; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_COMPRESSION; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_ENDPOINT; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_HEADERS; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_PROTOCOL; +import static datadog.trace.api.config.OtlpConfig.OTLP_PROFILES_TIMEOUT; import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_COMPRESSION; import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_ENDPOINT; import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_HEADERS; @@ -40,6 +46,8 @@ import static datadog.trace.api.config.OtlpConfig.OTLP_TRACES_TIMEOUT; import static datadog.trace.api.config.OtlpConfig.TRACE_OTEL_ENABLED; import static datadog.trace.api.config.OtlpConfig.TRACE_OTEL_EXPORTER; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED; +import static datadog.trace.api.config.ProfilingConfig.PROFILING_OTLP_ENABLED_DEFAULT; import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_ENABLED; import static datadog.trace.api.config.TraceInstrumentationConfig.TRACE_EXTENSIONS_PATH; import static datadog.trace.api.config.TracerConfig.REQUEST_HEADER_TAGS; @@ -104,11 +112,17 @@ public ConfigOrigin origin() { OtelEnvironmentConfigSource(Properties datadogConfigFile) { this.datadogConfigFile = datadogConfigFile; - this.enabled = traceOtelEnabled() || metricsOtelEnabled() || logsOtelEnabled(); + boolean otelEnabled = traceOtelEnabled() || metricsOtelEnabled() || logsOtelEnabled(); + boolean profilesEnabled = profilingOtlpEnabled(); + this.enabled = otelEnabled || profilesEnabled; - if (enabled) { + if (otelEnabled) { setupOtelEnvironment(); } + // OTLP profiling only maps the profiles exporter keys, not the general OTel environment + if (profilesEnabled) { + setupProfilesOtelEnvironment(); + } } private void setupOtelEnvironment() { @@ -266,6 +280,25 @@ private void setupLogsOtelEnvironment() { } } + // OTLP profile export is enabled by dd.profiling.otlp.enabled, not by an OTel exporter setting + private void setupProfilesOtelEnvironment() { + capture( + OTLP_PROFILES_HEADERS, + getOtelOtlpProperty("profiles", "headers", "dd." + OTLP_PROFILES_HEADERS)); + capture( + OTLP_PROFILES_PROTOCOL, + getOtelOtlpProperty("profiles", "protocol", "dd." + OTLP_PROFILES_PROTOCOL)); + capture( + OTLP_PROFILES_COMPRESSION, + getOtelOtlpProperty("profiles", "compression", "dd." + OTLP_PROFILES_COMPRESSION)); + capture( + OTLP_PROFILES_TIMEOUT, + getOtelOtlpProperty("profiles", "timeout", "dd." + OTLP_PROFILES_TIMEOUT)); + capture( + OTLP_PROFILES_ENDPOINT, + getOtelOtlpProperty("profiles", "endpoint", "dd." + OTLP_PROFILES_ENDPOINT)); + } + private boolean traceOtelEnabled() { String enabled = getDatadogProperty("dd." + TRACE_OTEL_ENABLED); if (null != enabled) { @@ -284,6 +317,15 @@ private boolean metricsOtelEnabled() { } } + private boolean profilingOtlpEnabled() { + String enabled = getDatadogProperty("dd." + PROFILING_OTLP_ENABLED); + if (null != enabled) { + return Boolean.parseBoolean(enabled); + } else { + return PROFILING_OTLP_ENABLED_DEFAULT; + } + } + private boolean logsOtelEnabled() { String enabled = getDatadogProperty("dd." + LOGS_OTEL_ENABLED); if (null != enabled) { @@ -355,6 +397,12 @@ private String getOtelOtlpProperty(String signal, String subkey, String ddSysPro && !"grpc".equalsIgnoreCase(otelEnvironment.get(OTLP_TRACES_PROTOCOL))) { otelValue = otelValue + (otelValue.endsWith("/") ? "v1/traces" : "/v1/traces"); } + if ("profiles".equals(signal) && !"grpc".equalsIgnoreCase(profilesProtocol())) { + otelValue = + otelValue + + (otelValue.endsWith("/") ? "" : "/") + + DEFAULT_OTLP_HTTP_PROFILES_ENDPOINT; + } } } if (null == otelValue) { @@ -371,6 +419,12 @@ private String getOtelOtlpProperty(String signal, String subkey, String ddSysPro return otelValue; } + /** The profiles protocol from the OTel environment, else from the Datadog setting that won. */ + private String profilesProtocol() { + String protocol = otelEnvironment.get(OTLP_PROFILES_PROTOCOL); + return protocol != null ? protocol : getDatadogProperty("dd." + OTLP_PROFILES_PROTOCOL); + } + /** * Gets a Datadog property. *