counts = new ArrayList<>(bucketCounts.length);
+ for (long count : bucketCounts) {
+ counts.add((double) count);
+ }
+
+ double sumSeconds = sumNanos / NANOS_PER_SECOND;
+ double minSeconds = histogram.isEmpty() ? 0d : histogram.getMinValue() / NANOS_PER_SECOND;
+ double maxSeconds = histogram.isEmpty() ? 0d : histogram.getMaxValue() / NANOS_PER_SECOND;
+
+ return new OtlpHistogramPoint(
+ histogram.getCount(), EXPLICIT_BOUNDS, counts, sumSeconds, minSeconds, maxSeconds);
+ }
+}
diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java
new file mode 100644
index 00000000000..98256d61d4f
--- /dev/null
+++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java
@@ -0,0 +1,239 @@
+package datadog.trace.common.metrics;
+
+import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.HISTOGRAM;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.LONG_ATTRIBUTE;
+import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE;
+import static datadog.trace.core.otlp.common.OtlpCommonProto.I64_WIRE_TYPE;
+import static datadog.trace.core.otlp.common.OtlpCommonProto.LEN_WIRE_TYPE;
+import static datadog.trace.core.otlp.common.OtlpCommonProto.writeAttribute;
+import static datadog.trace.core.otlp.common.OtlpCommonProto.writeI64;
+import static datadog.trace.core.otlp.common.OtlpCommonProto.writeTag;
+import static datadog.trace.core.otlp.common.OtlpResourceProto.RESOURCE_MESSAGE;
+import static datadog.trace.core.otlp.metrics.OtlpMetricsProto.recordDataPointMessage;
+import static datadog.trace.core.otlp.metrics.OtlpMetricsProto.recordMetricMessage;
+import static datadog.trace.core.otlp.metrics.OtlpMetricsProto.recordScopedMetricsMessage;
+
+import datadog.communication.serialization.GrowableBuffer;
+import datadog.metrics.api.Histogram;
+import datadog.trace.api.Config;
+import datadog.trace.bootstrap.instrumentation.api.UTF8BytesString;
+import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
+import datadog.trace.bootstrap.otel.metrics.OtelInstrumentDescriptor;
+import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint;
+import datadog.trace.core.otlp.common.OtlpGrpcSender;
+import datadog.trace.core.otlp.common.OtlpHttpSender;
+import datadog.trace.core.otlp.common.OtlpProtoBuffer;
+import datadog.trace.core.otlp.common.OtlpSender;
+import javax.annotation.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * A {@link MetricWriter} that exports the existing client-side trace (RED) stats as a single
+ * vendor-neutral OTLP delta-temporality histogram named {@code traces.span.sdk.metrics.duration}
+ * (unit {@code s}).
+ *
+ * This is the parallel-to-{@link SerializingMetricWriter} OTLP export path. It hangs off the
+ * same in-memory aggregation ({@link ClientStatsAggregator} / {@link Aggregator}) and consumes the
+ * same {@link AggregateEntry} stream; only the wire encoding and transport differ. Native msgpack
+ * stats and OTLP export are mutually exclusive (selected at the factory).
+ *
+ *
Assembly mirrors {@code OtlpMetricsProtoCollector}
+ */
+public final class OtlpStatsMetricWriter implements MetricWriter {
+ private static final Logger log = LoggerFactory.getLogger(OtlpStatsMetricWriter.class);
+
+ static final String METRIC_NAME = "traces.span.sdk.metrics.duration";
+ static final String METRIC_UNIT = "s";
+
+ private static final OtelInstrumentDescriptor METRIC_DESCRIPTOR =
+ new OtelInstrumentDescriptor(METRIC_NAME, HISTOGRAM, false, null, METRIC_UNIT);
+ private static final OtelInstrumentationScope SCOPE =
+ new OtelInstrumentationScope("datadog.trace.metrics", null, null);
+
+ private static final int DP_START_TIME_FIELD = 2;
+ private static final int DP_TIME_FIELD = 3;
+ private static final int DP_ATTRIBUTES_FIELD = 9;
+
+ private static final String SPAN_NAME = "span.name";
+ private static final String SPAN_KIND = "span.kind";
+ private static final String HTTP_REQUEST_METHOD = "http.request.method";
+ private static final String HTTP_RESPONSE_STATUS_CODE = "http.response.status_code";
+ private static final String HTTP_ROUTE = "http.route";
+ private static final String RPC_RESPONSE_STATUS_CODE = "rpc.response.status_code";
+ private static final String STATUS_CODE = "status.code";
+ private static final String STATUS_CODE_ERROR = "ERROR";
+ private static final String DATADOG_OPERATION_NAME = "datadog.operation.name";
+ private static final String DATADOG_SPAN_TYPE = "datadog.span.type";
+ private static final String DATADOG_SPAN_TOP_LEVEL = "datadog.span.top_level";
+ private static final String DATADOG_ORIGIN = "datadog.origin";
+ private static final String SYNTHETICS_ORIGIN = "synthetics";
+
+ @Nullable private final OtlpSender sender;
+ private final boolean otelSemanticsMode;
+
+ // Need a temporary buffer to know what size to write for the final protobuf buffer
+ private final GrowableBuffer buf = new GrowableBuffer(512);
+ private final OtlpProtoBuffer protobuf = new OtlpProtoBuffer(8192);
+
+ private long startNanos;
+ private long endNanos;
+
+ private int payloadBytes;
+ private int scopedBytes;
+ private int metricBytes;
+
+ public OtlpStatsMetricWriter(Config config) {
+ this(createSender(config), config.isTraceOtelSemanticsEnabled());
+ }
+
+ // visible for testing: lets tests inject a capturing sender to decode the emitted protobuf and
+ // control the semantics mode
+ OtlpStatsMetricWriter(@Nullable OtlpSender sender, boolean otelSemanticsMode) {
+ this.sender = sender;
+ this.otelSemanticsMode = otelSemanticsMode;
+ }
+
+ @Nullable
+ private static OtlpSender createSender(Config config) {
+ // mirrors OtlpMetricsService's protocol-based sender selection
+ switch (config.getOtlpMetricsProtocol()) {
+ case GRPC:
+ return new OtlpGrpcSender(
+ config.getOtlpMetricsEndpoint(),
+ "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export",
+ config.getOtlpMetricsHeaders(),
+ config.getOtlpMetricsTimeout(),
+ config.getOtlpMetricsCompression());
+ case HTTP_PROTOBUF:
+ return new OtlpHttpSender(
+ config.getOtlpMetricsEndpoint(),
+ "/v1/metrics",
+ config.getOtlpMetricsHeaders(),
+ config.getOtlpMetricsTimeout(),
+ config.getOtlpMetricsCompression());
+ default:
+ // HTTP_JSON has no protobuf-free encoder yet; JSON transport is deferred per the plan.
+ log.warn(
+ "OTLP trace metrics export disabled: unsupported metrics protocol {}. "
+ + "Set OTEL_EXPORTER_OTLP_METRICS_PROTOCOL to grpc or http/protobuf.",
+ config.getOtlpMetricsProtocol());
+ return null;
+ }
+ }
+
+ @Override
+ public void startBucket(int metricCount, long start, long duration) {
+ // start/duration arrive as epoch nanos / interval nanos (see Aggregator#report)
+ this.startNanos = start;
+ this.endNanos = start + duration;
+ this.payloadBytes = 0;
+ this.scopedBytes = 0;
+ this.metricBytes = 0;
+ }
+
+ @Override
+ public void add(AggregateEntry entry) {
+ Histogram okLatencies = entry.getOkLatencies();
+ if (!okLatencies.isEmpty()) {
+ addDataPoint(entry, okLatencies, false);
+ }
+
+ Histogram errorLatencies = entry.getErrorLatencies();
+ if (errorLatencies != null && !errorLatencies.isEmpty()) {
+ addDataPoint(entry, errorLatencies, true);
+ }
+ }
+
+ private void addDataPoint(AggregateEntry entry, Histogram latencies, boolean error) {
+ writeDataPointAttributes(entry, error);
+ writeTag(buf, DP_START_TIME_FIELD, I64_WIRE_TYPE);
+ writeI64(buf, startNanos);
+ writeTag(buf, DP_TIME_FIELD, I64_WIRE_TYPE);
+ writeI64(buf, endNanos);
+ long sumNanos = error ? entry.getErrorDuration() : entry.getOkDuration();
+ OtlpHistogramPoint point = OtlpStatsHistogramBuckets.toHistogramPoint(latencies, sumNanos);
+ metricBytes += recordDataPointMessage(buf, point, protobuf);
+ }
+
+ private void writeDataPointAttributes(AggregateEntry entry, boolean error) {
+ if (error) {
+ writeStringAttribute(STATUS_CODE, STATUS_CODE_ERROR);
+ }
+ // OTel semconv attrs are emitted in both modes
+ writeStringAttribute(SPAN_NAME, entry.getResource());
+ writeStringAttribute(SPAN_KIND, entry.getSpanKind());
+ if (entry.hasHttpMethod()) {
+ writeStringAttribute(HTTP_REQUEST_METHOD, entry.getHttpMethod());
+ }
+ if (entry.getHttpStatusCode() != 0) {
+ writeLongAttribute(HTTP_RESPONSE_STATUS_CODE, entry.getHttpStatusCode());
+ }
+ if (entry.hasHttpEndpoint()) {
+ writeStringAttribute(HTTP_ROUTE, entry.getHttpEndpoint());
+ }
+ if (entry.hasGrpcStatusCode()) {
+ writeStringAttribute(RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode());
+ }
+ // Default (Datadog) mode: emit datadog.* per-point attributes
+ if (!otelSemanticsMode) {
+ writeStringAttribute(DATADOG_OPERATION_NAME, entry.getOperationName());
+ writeStringAttribute(DATADOG_SPAN_TYPE, entry.getType());
+ writeLongAttribute(
+ DATADOG_SPAN_TOP_LEVEL, entry.getTopLevelCount() == entry.getHitCount() ? 1L : 0L);
+ if (entry.isSynthetics()) {
+ writeStringAttribute(DATADOG_ORIGIN, SYNTHETICS_ORIGIN);
+ }
+ }
+ }
+
+ private void writeStringAttribute(String key, @Nullable UTF8BytesString value) {
+ if (value != null) {
+ writeStringAttribute(key, value.toString());
+ }
+ }
+
+ private void writeStringAttribute(String key, String value) {
+ writeTag(buf, DP_ATTRIBUTES_FIELD, LEN_WIRE_TYPE);
+ writeAttribute(buf, STRING_ATTRIBUTE, key, value);
+ }
+
+ private void writeLongAttribute(String key, long value) {
+ writeTag(buf, DP_ATTRIBUTES_FIELD, LEN_WIRE_TYPE);
+ writeAttribute(buf, LONG_ATTRIBUTE, key, value);
+ }
+
+ @Override
+ public void finishBucket() {
+ try {
+ if (metricBytes > 0) {
+ // trace stats histograms are inherently per-interval deltas (buckets are cleared after
+ // every flush), so always encode DELTA regardless of the temporality preference
+ scopedBytes += recordMetricMessage(buf, METRIC_DESCRIPTOR, metricBytes, protobuf, true);
+ }
+ if (scopedBytes > 0) {
+ payloadBytes += recordScopedMetricsMessage(buf, SCOPE, scopedBytes, protobuf);
+ }
+ if (payloadBytes == 0) {
+ return;
+ }
+ payloadBytes += protobuf.recordMessage(RESOURCE_MESSAGE);
+ protobuf.recordMessage(buf, 1, payloadBytes);
+
+ if (sender != null) {
+ sender.send(protobuf.toPayload());
+ }
+ } finally {
+ reset();
+ }
+ }
+
+ @Override
+ public void reset() {
+ buf.reset();
+ protobuf.reset();
+ payloadBytes = 0;
+ scopedBytes = 0;
+ metricBytes = 0;
+ }
+}
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java
index 751649940ba..a92c0826340 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java
@@ -60,6 +60,20 @@ public static int recordMetricMessage(
OtelInstrumentDescriptor descriptor,
int nestedDataPointBytes,
OtlpProtoBuffer protobuf) {
+ return recordMetricMessage(buf, descriptor, nestedDataPointBytes, protobuf, false);
+ }
+
+ /**
+ * Records a metric message after its nested data point messages have been recorded. When {@code
+ * forceDelta} is {@code true}, a histogram is encoded as DELTA regardless of the configured
+ * temporality preference (for sources whose data points are inherently per-interval deltas).
+ */
+ public static int recordMetricMessage(
+ GrowableBuffer buf,
+ OtelInstrumentDescriptor descriptor,
+ int nestedDataPointBytes,
+ OtlpProtoBuffer protobuf,
+ boolean forceDelta) {
writeTag(buf, 1, LEN_WIRE_TYPE);
writeString(buf, descriptor.getName().getUtf8Bytes());
@@ -107,7 +121,7 @@ public static int recordMetricMessage(
writeTag(buf, 9, LEN_WIRE_TYPE);
writeVarInt(buf, nestedDataPointBytes + 2);
writeTag(buf, 2, VARINT_WIRE_TYPE);
- writeVarInt(buf, HISTOGRAM_TEMPORALITY);
+ writeVarInt(buf, forceDelta ? TEMPORALITY_DELTA : HISTOGRAM_TEMPORALITY);
break;
default:
throw new IllegalArgumentException("Unknown instrument type: " + descriptor.getType());
diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java
new file mode 100644
index 00000000000..d37cde2d92a
--- /dev/null
+++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java
@@ -0,0 +1,145 @@
+package datadog.trace.common.metrics;
+
+import static datadog.trace.common.metrics.OtlpStatsHistogramBuckets.BOUNDS_SECONDS;
+import static datadog.trace.common.metrics.OtlpStatsHistogramBuckets.EXPLICIT_BOUNDS;
+import static datadog.trace.common.metrics.OtlpStatsHistogramBuckets.bucketIndex;
+import static datadog.trace.common.metrics.OtlpStatsHistogramBuckets.toHistogramPoint;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import datadog.metrics.api.Histogram;
+import datadog.metrics.api.Histograms;
+import datadog.metrics.impl.DDSketchHistograms;
+import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+/**
+ * Unit tests for {@link OtlpStatsHistogramBuckets}, the helper that re-bins a nanosecond-valued
+ * DDSketch onto the fixed OTLP explicit-bounds histogram layout (in seconds).
+ */
+class OtlpStatsHistogramBucketsTest {
+
+ private static final double NANOS_PER_SECOND = 1_000_000_000d;
+
+ // Tolerance for double assertions on exactly-computed values (e.g. sum from sumNanos, integer
+ // counts). DDSketch-derived values like min/max use looser tolerances inline, since the sketch
+ // only preserves values to within its relative accuracy.
+ private static final double EPS = 1e-9;
+
+ @BeforeAll
+ static void registerHistogramFactory() {
+ Histograms.register(DDSketchHistograms.FACTORY);
+ }
+
+ // ── bucketIndex ──────────────────────────────────────────────────────────
+
+ static Stream boundsWithIndex() {
+ return IntStream.range(0, BOUNDS_SECONDS.length)
+ .mapToObj(i -> Arguments.of(i, BOUNDS_SECONDS[i]));
+ }
+
+ @ParameterizedTest(name = "value exactly on BOUNDS_SECONDS[{0}]={1} returns index {0}")
+ @MethodSource("boundsWithIndex")
+ void valueExactlyOnBoundReturnsThatIndex(int index, double bound) {
+ // <= semantics: a value exactly on a bound falls in that bound's bucket.
+ assertEquals(index, bucketIndex(bound));
+ }
+
+ @Test
+ void valueJustAboveSmallBoundFallsInNextBucket() {
+ // 0.002 is the first bound; a value just above it but below the second bound (0.004)
+ // must roll into the next index.
+ assertEquals(0, bucketIndex(0.002));
+ assertEquals(1, bucketIndex(0.0020001));
+ assertEquals(1, bucketIndex(0.004));
+ }
+
+ @Test
+ void valueAboveLargestBoundIsOverflow() {
+ // > 15s overflows to the final (16th) index.
+ assertEquals(BOUNDS_SECONDS.length, bucketIndex(15.0000001));
+ assertEquals(BOUNDS_SECONDS.length, bucketIndex(100.0));
+ // exactly 15 (the largest non-overflow bound) is the last non-overflow index.
+ assertEquals(BOUNDS_SECONDS.length - 1, bucketIndex(15.0));
+ }
+
+ // ── EXPLICIT_BOUNDS layout ────────────────────────────────────────────────
+
+ @Test
+ void explicitBoundsHas17EntriesEndingInInfinity() {
+ assertEquals(17, EXPLICIT_BOUNDS.size());
+ assertEquals(Double.POSITIVE_INFINITY, EXPLICIT_BOUNDS.get(EXPLICIT_BOUNDS.size() - 1));
+ }
+
+ // ── toHistogramPoint ──────────────────────────────────────────────────────
+
+ @Test
+ void toHistogramPointSummary() {
+ Histogram h = Histogram.newHistogram();
+ // 1ns and 1ms are below the smallest bound (0.002s) and must land in bucket 0; 100ms → bucket
+ // 6, 3s → bucket 13.
+ long[] samplesNanos = {
+ 1L,
+ (long) (0.001 * NANOS_PER_SECOND),
+ (long) (0.1 * NANOS_PER_SECOND),
+ (long) (3.0 * NANOS_PER_SECOND)
+ };
+ for (long s : samplesNanos) {
+ h.accept(s);
+ }
+
+ // Use a sumNanos that deliberately differs from the sketch's implied sum to prove the
+ // returned sum comes from the ARGUMENT, not the sketch.
+ long sumNanos = 42L * (long) NANOS_PER_SECOND;
+ OtlpHistogramPoint point = toHistogramPoint(h, sumNanos);
+
+ // 17 bucket counts (the EXPLICIT_BOUNDS layout itself is covered by its own test).
+ assertEquals(17, point.bucketCounts.size());
+
+ // total count == number of samples.
+ assertEquals(samplesNanos.length, (long) point.count);
+
+ // max is the 3s sample (CollapsingLowestDenseStore collapses the LOWEST bins, so the top is
+ // preserved accurately). The exact 1ns min is NOT recoverable: over this wide value range the
+ // lowest bins collapse, so getMinValue reports the collapsed bin (sub-2ms here), not 1ns. The
+ // tiny sample isn't lost though — that's proven by the count and bucket-0 assertions below.
+ // DDSketch is relative-accuracy, so min/max use loose tolerances.
+ assertTrue(
+ point.min > 0 && point.min <= BOUNDS_SECONDS[0], "min collapses into bucket 0 range");
+ assertEquals(3.0, point.max, 1e-2);
+
+ // sum equals the sumNanos ARGUMENT converted to seconds, not the sketch sum.
+ assertEquals(42.0, point.sum, EPS);
+
+ long bucketTotal = 0;
+ for (int i = 0; i < point.bucketCounts.size(); i++) {
+ long c = (long) point.bucketCounts.get(i).doubleValue();
+ bucketTotal += c;
+ if (i == 0) {
+ assertEquals(2L, c, "1ns + 1ms both land in bucket 0 (<= 0.002s)");
+ }
+ }
+ assertEquals(samplesNanos.length, bucketTotal);
+ }
+
+ @Test
+ void emptyHistogramHasZeroMinMaxAndCount() {
+ Histogram h = Histogram.newHistogram();
+ OtlpHistogramPoint point = toHistogramPoint(h, 0L);
+
+ assertEquals(0.0, point.count, EPS);
+ assertEquals(0.0, point.min, EPS);
+ assertEquals(0.0, point.max, EPS);
+ long bucketTotal = 0;
+ for (double c : point.bucketCounts) {
+ bucketTotal += (long) c;
+ }
+ assertEquals(0L, bucketTotal);
+ }
+}
diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java
new file mode 100644
index 00000000000..31ed707cabd
--- /dev/null
+++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java
@@ -0,0 +1,477 @@
+package datadog.trace.common.metrics;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+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.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+
+import com.google.protobuf.CodedInputStream;
+import com.google.protobuf.WireFormat;
+import datadog.metrics.api.Histograms;
+import datadog.metrics.impl.DDSketchHistograms;
+import datadog.trace.core.otlp.common.OtlpPayload;
+import datadog.trace.core.otlp.common.OtlpSender;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import javax.annotation.Nullable;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+/**
+ * Tests for {@link OtlpStatsMetricWriter}. Drives the writer through {@code startBucket} → {@code
+ * add} → {@code finishBucket} with a capturing {@link OtlpSender}, then decodes the emitted
+ * protobuf {@code ExportMetricsServiceRequest} ({@code MetricsData}) using protobuf's {@link
+ * CodedInputStream}, reusing the decode idioms from {@code OtlpMetricsProtoTest}.
+ *
+ * Wire layout (OTLP metrics proto):
+ *
+ *
+ * MetricsData { ResourceMetrics resource_metrics = 1; }
+ * ResourceMetrics { Resource resource = 1; ScopeMetrics scope_metrics = 2; }
+ * ScopeMetrics { InstrumentationScope scope = 1; Metric metrics = 2; }
+ * Metric { string name = 1; string unit = 3; Histogram histogram = 9; }
+ * Histogram { HistogramDataPoint data_points = 1; AggregationTemporality = 2; }
+ * HistogramDataPoint { fixed64 start = 2; fixed64 time = 3; fixed64 count = 4; double sum = 5;
+ * fixed64 bucket_counts = 6; double explicit_bounds = 7;
+ * KeyValue attributes = 9; double min = 11; double max = 12; }
+ *
+ */
+class OtlpStatsMetricWriterTest {
+
+ private static final int TEMPORALITY_DELTA = 1;
+ private static final long BUCKET_START = SECONDS.toNanos(1_700_000_000L);
+ private static final long BUCKET_DURATION = SECONDS.toNanos(10);
+
+ @BeforeAll
+ static void registerHistogramFactory() {
+ Histograms.register(DDSketchHistograms.FACTORY);
+ }
+
+ // ── capturing sender ──────────────────────────────────────────────────────
+
+ private static final class CapturingSender implements OtlpSender {
+ int sendCount;
+ byte[] lastPayload;
+
+ @Override
+ public void send(OtlpPayload payload) {
+ sendCount++;
+ java.nio.ByteBuffer content = payload.getContent();
+ byte[] bytes = new byte[content.remaining()];
+ content.get(bytes);
+ lastPayload = bytes;
+ }
+
+ @Override
+ public void shutdown() {}
+ }
+
+ // ── entry builders ──────────────────────────────────────────────────────
+
+ private static AggregateEntry entry(
+ String resource,
+ boolean synthetic,
+ int httpStatusCode,
+ @Nullable String httpMethod,
+ @Nullable String httpEndpoint,
+ @Nullable String grpcStatusCode) {
+ return AggregateEntryTestUtils.of(
+ resource,
+ "web",
+ "servlet.request",
+ null,
+ "web",
+ httpStatusCode,
+ synthetic,
+ true,
+ "server",
+ null,
+ httpMethod,
+ httpEndpoint,
+ grpcStatusCode);
+ }
+
+ /** Build an entry and record {@code hits} ok durations of {@code durationNanos} each. */
+ private static AggregateEntry okEntry(long durationNanos, int hits) {
+ AggregateEntry e = entry("GET /users", false, 0, null, null, null);
+ for (int i = 0; i < hits; i++) {
+ e.recordOneDuration(durationNanos);
+ }
+ return e;
+ }
+
+ // ── decode helpers (adapted from OtlpMetricsProtoTest) ──────────────────────
+
+ /**
+ * A decoded histogram data point. Only the fields this test asserts on are decoded: the window
+ * timestamps, the total count, and the attributes. Per-bucket contents (bucket_counts,
+ * explicit_bounds), sum, min, and max are intentionally not decoded here.
+ */
+ private static final class DataPoint {
+ long start;
+ long end;
+ long count;
+ final Map attributes = new HashMap<>();
+ }
+
+ /** A decoded metric: name, unit, temporality, and its histogram data points. */
+ private static final class DecodedMetric {
+ String name;
+ String unit;
+ int temporality = -1;
+ final List dataPoints = new ArrayList<>();
+ }
+
+ /** Decodes a full {@code MetricsData} payload into the single histogram metric it carries. */
+ private static DecodedMetric decode(byte[] payload) throws IOException {
+ CodedInputStream metricsData = CodedInputStream.newInstance(payload);
+ int metricsTag = metricsData.readTag();
+ assertEquals(1, WireFormat.getTagFieldNumber(metricsTag), "MetricsData.resource_metrics = 1");
+ CodedInputStream resourceMetrics = metricsData.readBytes().newCodedInput();
+ assertTrue(metricsData.isAtEnd(), "expected exactly one ResourceMetrics");
+
+ DecodedMetric metric = null;
+ while (!resourceMetrics.isAtEnd()) {
+ int tag = resourceMetrics.readTag();
+ int field = WireFormat.getTagFieldNumber(tag);
+ if (field == 2) { // ScopeMetrics
+ metric = parseScopeMetrics(resourceMetrics.readBytes().newCodedInput());
+ } else {
+ resourceMetrics.skipField(tag); // Resource (field 1) etc.
+ }
+ }
+ assertNotNull(metric, "no ScopeMetrics found");
+ return metric;
+ }
+
+ private static DecodedMetric parseScopeMetrics(CodedInputStream scopeMetrics) throws IOException {
+ DecodedMetric metric = null;
+ while (!scopeMetrics.isAtEnd()) {
+ int tag = scopeMetrics.readTag();
+ if (WireFormat.getTagFieldNumber(tag) == 2) { // Metric
+ metric = parseMetric(scopeMetrics.readBytes().newCodedInput());
+ } else {
+ scopeMetrics.skipField(tag);
+ }
+ }
+ return metric;
+ }
+
+ private static DecodedMetric parseMetric(CodedInputStream m) throws IOException {
+ DecodedMetric metric = new DecodedMetric();
+ while (!m.isAtEnd()) {
+ int tag = m.readTag();
+ switch (WireFormat.getTagFieldNumber(tag)) {
+ case 1:
+ metric.name = m.readString();
+ break;
+ case 3:
+ metric.unit = m.readString();
+ break;
+ case 9: // Histogram
+ parseHistogram(m.readBytes().newCodedInput(), metric);
+ break;
+ default:
+ m.skipField(tag);
+ }
+ }
+ return metric;
+ }
+
+ private static void parseHistogram(CodedInputStream h, DecodedMetric metric) throws IOException {
+ while (!h.isAtEnd()) {
+ int tag = h.readTag();
+ switch (WireFormat.getTagFieldNumber(tag)) {
+ case 1: // HistogramDataPoint (repeated)
+ metric.dataPoints.add(parseDataPoint(h.readBytes().newCodedInput()));
+ break;
+ case 2: // aggregation_temporality
+ metric.temporality = h.readEnum();
+ break;
+ default:
+ h.skipField(tag);
+ }
+ }
+ }
+
+ private static DataPoint parseDataPoint(CodedInputStream dp) throws IOException {
+ DataPoint p = new DataPoint();
+ while (!dp.isAtEnd()) {
+ int tag = dp.readTag();
+ switch (WireFormat.getTagFieldNumber(tag)) {
+ case 2:
+ p.start = dp.readFixed64();
+ break;
+ case 3:
+ p.end = dp.readFixed64();
+ break;
+ case 4:
+ p.count = dp.readFixed64();
+ break;
+ case 9: // attributes (KeyValue)
+ readKeyValue(dp.readBytes().newCodedInput(), p.attributes);
+ break;
+ default: // sum, bucket_counts, explicit_bounds, min, max — not asserted here
+ dp.skipField(tag);
+ }
+ }
+ return p;
+ }
+
+ /**
+ * Reads a {@code KeyValue} into {@code out}: key (field 1) → value. Value is an {@code AnyValue}
+ * (field 2); we decode string (field 1) and int (field 3) variants used by this writer.
+ */
+ private static void readKeyValue(CodedInputStream kv, Map out)
+ throws IOException {
+ String key = null;
+ Object value = null;
+ while (!kv.isAtEnd()) {
+ int tag = kv.readTag();
+ switch (WireFormat.getTagFieldNumber(tag)) {
+ case 1:
+ key = kv.readString();
+ break;
+ case 2: // AnyValue
+ value = readAnyValue(kv.readBytes().newCodedInput());
+ break;
+ default:
+ kv.skipField(tag);
+ }
+ }
+ if (key != null) {
+ out.put(key, value);
+ }
+ }
+
+ private static Object readAnyValue(CodedInputStream any) throws IOException {
+ Object value = null;
+ while (!any.isAtEnd()) {
+ int tag = any.readTag();
+ switch (WireFormat.getTagFieldNumber(tag)) {
+ case 1: // string_value
+ value = any.readString();
+ break;
+ case 3: // int_value
+ value = any.readInt64();
+ break;
+ default:
+ any.skipField(tag);
+ }
+ }
+ return value;
+ }
+
+ // ── writer driver ─────────────────────────────────────────────────────────
+
+ /**
+ * Drives the writer through one full {@code startBucket → add → finishBucket} cycle for {@code
+ * entry} over the fixed {@link #BUCKET_START}/{@link #BUCKET_DURATION} window, asserts that
+ * exactly one payload was sent, and returns the decoded metric.
+ */
+ private static DecodedMetric writeAndDecode(boolean otelSemanticsMode, AggregateEntry entry)
+ throws IOException {
+ CapturingSender sender = new CapturingSender();
+ OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, otelSemanticsMode);
+ writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
+ writer.add(entry);
+ writer.finishBucket();
+ assertEquals(1, sender.sendCount, "exactly one payload sent");
+ return decode(sender.lastPayload);
+ }
+
+ // ── test cases ──────────────────────────────────────────────────────────
+
+ @Test
+ void okOnlyEntryProducesExactlyOneDataPoint() throws IOException {
+ DecodedMetric metric = writeAndDecode(false, okEntry(SECONDS.toNanos(1), 3));
+
+ assertEquals("traces.span.sdk.metrics.duration", metric.name);
+ assertEquals("s", metric.unit);
+ assertEquals(TEMPORALITY_DELTA, metric.temporality, "histogram must be delta temporality");
+ assertEquals(1, metric.dataPoints.size(), "ok-only → one data point");
+
+ DataPoint dp = metric.dataPoints.get(0);
+ assertEquals(BUCKET_START, dp.start, "start_time_unix_nano == startBucket start");
+ assertEquals(BUCKET_START + BUCKET_DURATION, dp.end, "time_unix_nano == start + duration");
+ assertEquals(3L, dp.count);
+ assertFalse(dp.attributes.containsKey("status.code"), "ok point carries no status.code");
+ }
+
+ @Test
+ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
+ AggregateEntry e = entry("GET /users", false, 0, null, null, null);
+ e.recordOneDuration(SECONDS.toNanos(1)); // ok
+ e.recordOneDuration(SECONDS.toNanos(2)); // ok
+ e.recordOneDuration(SECONDS.toNanos(3) | AggregateEntry.ERROR_TAG); // error
+
+ DecodedMetric metric = writeAndDecode(false, e);
+ assertEquals(2, metric.dataPoints.size(), "ok+error → two data points");
+
+ long okCount = 0;
+ long errorCount = 0;
+ DataPoint errorPoint = null;
+ DataPoint okPoint = null;
+ for (DataPoint dp : metric.dataPoints) {
+ if ("ERROR".equals(dp.attributes.get("status.code"))) {
+ errorPoint = dp;
+ errorCount = dp.count;
+ } else {
+ okPoint = dp;
+ okCount = dp.count;
+ }
+ }
+ assertNotNull(errorPoint, "one data point must carry status.code=ERROR");
+ assertNotNull(okPoint, "one data point must omit status.code");
+ assertEquals(e.getOkLatencies().getCount(), (double) okCount, 1e-9);
+ assertEquals(e.getErrorLatencies().getCount(), (double) errorCount, 1e-9);
+ }
+
+ @Test
+ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOException {
+ CapturingSender sender = new CapturingSender();
+ OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false);
+
+ // Bucket 1: the entry sees an error, so its error histogram is allocated and emits a point.
+ AggregateEntry e = entry("GET /users", false, 0, null, null, null);
+ e.recordOneDuration(SECONDS.toNanos(1)); // ok
+ e.recordOneDuration(SECONDS.toNanos(3) | AggregateEntry.ERROR_TAG); // error
+
+ writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
+ writer.add(e);
+ writer.finishBucket();
+ DecodedMetric bucket1 = decode(sender.lastPayload);
+ assertEquals(2, bucket1.dataPoints.size(), "bucket with an error → ok+error data points");
+ assertTrue(
+ bucket1.dataPoints.stream()
+ .anyMatch(dp -> "ERROR".equals(dp.attributes.get("status.code"))),
+ "bucket 1 must carry a status.code=ERROR point");
+
+ // Bucket 2: same entry, reset then only OK hits. errorLatencies survives clear() (cleared, not
+ // nulled), so a non-null-but-empty histogram must NOT emit a phantom zero-count error series.
+ e.clear();
+ e.recordOneDuration(SECONDS.toNanos(2)); // ok only
+
+ writer.startBucket(2, BUCKET_START + BUCKET_DURATION, BUCKET_DURATION);
+ writer.add(e);
+ writer.finishBucket();
+ DecodedMetric bucket2 = decode(sender.lastPayload);
+ assertEquals(1, bucket2.dataPoints.size(), "ok-only bucket → exactly one data point");
+ assertFalse(
+ bucket2.dataPoints.get(0).attributes.containsKey("status.code"),
+ "recovered entry must not emit a lingering status.code=ERROR series");
+ }
+
+ @Test
+ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
+ AggregateEntry e = entry("GET /users/{id}", false, 200, "GET", "/users/{id}", "0");
+ e.recordOneDuration(SECONDS.toNanos(1));
+
+ DecodedMetric metric = writeAndDecode(false, e);
+ assertEquals(1, metric.dataPoints.size());
+ Map attrs = metric.dataPoints.get(0).attributes;
+
+ assertEquals("GET", attrs.get("http.request.method"));
+ assertEquals(200L, attrs.get("http.response.status_code"));
+ assertEquals("/users/{id}", attrs.get("http.route"));
+ assertEquals("0", attrs.get("rpc.response.status_code"));
+
+ // a bare entry has none of these
+ Map bareAttrs =
+ writeAndDecode(false, okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes;
+ assertFalse(bareAttrs.containsKey("http.request.method"));
+ assertFalse(bareAttrs.containsKey("http.response.status_code"));
+ assertFalse(bareAttrs.containsKey("http.route"));
+ assertFalse(bareAttrs.containsKey("rpc.response.status_code"));
+ }
+
+ @Test
+ void emptyBucketSendsNothing() {
+ CapturingSender sender = new CapturingSender();
+ OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false);
+
+ writer.startBucket(0, BUCKET_START, BUCKET_DURATION);
+ writer.finishBucket(); // no add()
+
+ assertEquals(0, sender.sendCount, "empty bucket must not invoke send");
+ assertNull(sender.lastPayload);
+ }
+
+ @Test
+ void nullSenderDoesNotThrowOnNonEmptyBucket() {
+ // mirrors the HTTP_JSON path where createSender(config) returns null.
+ OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(null, false);
+ writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
+ writer.add(okEntry(SECONDS.toNanos(1), 2));
+ try {
+ writer.finishBucket();
+ } catch (Exception ex) {
+ fail("finishBucket must not throw with a null sender, but threw: " + ex);
+ }
+ }
+
+ @Test
+ void defaultModeCarriesDatadogAttributes() throws IOException {
+ // use an entry where all hits are top-level: OR in TOP_LEVEL_TAG
+ AggregateEntry e = entry("servlet.request", false, 0, null, null, null);
+ e.recordOneDuration(SECONDS.toNanos(1) | AggregateEntry.TOP_LEVEL_TAG);
+
+ Map attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+ assertTrue(
+ attrs.containsKey("datadog.operation.name"), "operation name present in default mode");
+ assertTrue(attrs.containsKey("datadog.span.type"), "span type present in default mode");
+ assertTrue(
+ attrs.containsKey("datadog.span.top_level"), "span top-level present in default mode");
+ assertEquals(1L, attrs.get("datadog.span.top_level"), "all hits top-level → 1");
+ // OTel-semconv attrs are present in both modes
+ assertTrue(attrs.containsKey("span.name"), "span.name present in both modes");
+ // datadog.origin presence/absence is covered by defaultModeEmitsSyntheticOrigin
+ }
+
+ /**
+ * In default mode a synthetic entry emits {@code datadog.origin = "synthetics"}; a non-synthetic
+ * entry omits the attribute. Origin has collapsed to a boolean {@code synthetic} flag upstream,
+ * so {@code "synthetics"} is the only origin value that can reach the writer.
+ */
+ @ParameterizedTest(name = "synthetic={0} → datadog.origin={1}")
+ @CsvSource(
+ nullValues = "NULL",
+ value = {"false, NULL", "true, synthetics"})
+ void defaultModeEmitsSyntheticOrigin(boolean synthetic, String expectedOrigin)
+ throws IOException {
+ AggregateEntry e = entry("servlet.request", synthetic, 0, null, null, null);
+ e.recordOneDuration(SECONDS.toNanos(1));
+
+ Map attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
+ if (expectedOrigin == null) {
+ assertFalse(attrs.containsKey("datadog.origin"), "non-synthetic → datadog.origin absent");
+ } else {
+ assertEquals(expectedOrigin, attrs.get("datadog.origin"), "synthetic → datadog.origin");
+ }
+ }
+
+ @Test
+ void otelSemanticsModeOmitsDatadogAttributes() throws IOException {
+ // otelSemanticsMode = true → datadog.* must be absent
+ Map attrs =
+ writeAndDecode(true, okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes;
+ assertFalse(
+ attrs.containsKey("datadog.operation.name"),
+ "operation name absent in otel-semantics mode");
+ assertFalse(attrs.containsKey("datadog.span.type"), "span type absent in otel-semantics mode");
+ assertFalse(
+ attrs.containsKey("datadog.span.top_level"),
+ "span top-level absent in otel-semantics mode");
+ assertFalse(attrs.containsKey("datadog.origin"), "origin absent in otel-semantics mode");
+ // OTel-semconv attrs must still be present
+ assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode");
+ }
+}