From 2d8a515b01504bff4840e9b65843c85e8059a7d0 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Tue, 23 Jun 2026 16:28:32 -0400 Subject: [PATCH 01/16] init --- .../trace/common/metrics/AggregateEntry.java | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java index 29ce3d0bea4..b18fce272a4 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java @@ -105,7 +105,8 @@ final class AggregateEntry extends Hashtable.Entry { private int errorCount; private int hitCount; private int topLevelCount; - private long duration; + private long okDuration; + private long errorDuration; /** * Field-bearing constructor. Package-private so {@link AggregateEntryTestUtils} can build @@ -155,11 +156,12 @@ AggregateEntry recordOneDuration(long tagAndDuration) { if ((tagAndDuration & ERROR_TAG) == ERROR_TAG) { tagAndDuration ^= ERROR_TAG; errorLatenciesForWrite().accept(tagAndDuration); + errorDuration += tagAndDuration; ++errorCount; } else { okLatencies.accept(tagAndDuration); + okDuration += tagAndDuration; } - duration += tagAndDuration; return this; } @@ -176,7 +178,15 @@ int getTopLevelCount() { } long getDuration() { - return duration; + return okDuration + errorDuration; + } + + long getOkDuration() { + return okDuration; + } + + long getErrorDuration() { + return errorDuration; } Histogram getOkLatencies() { @@ -214,7 +224,8 @@ void clear() { this.errorCount = 0; this.hitCount = 0; this.topLevelCount = 0; - this.duration = 0; + this.okDuration = 0; + this.errorDuration = 0; this.okLatencies.clear(); // errorLatencies stays null on entries that never errored. Only clear if it was allocated. if (this.errorLatencies != null) { From 5942fe0888436d376544ee85f41dd26db3ef29fa Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Tue, 16 Jun 2026 15:24:41 -0400 Subject: [PATCH 02/16] init --- .../common/metrics/OtlpHistogramBuckets.java | 79 +++++++ .../common/metrics/OtlpStatsMetricWriter.java | 218 ++++++++++++++++++ .../core/otlp/common/OtlpResourceProto.java | 6 + .../otlp/common/OtlpResourceProtoTest.java | 11 + 4 files changed, 314 insertions(+) create mode 100644 dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java create mode 100644 dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java new file mode 100644 index 00000000000..977bf74ead6 --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java @@ -0,0 +1,79 @@ +package datadog.trace.common.metrics; + +import datadog.metrics.api.Histogram; +import datadog.metrics.api.HistogramWithSum; +import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; + +/** + * Projects a client-side-stats {@link Histogram} (a DDSketch over span durations recorded in + * nanoseconds) onto the fixed explicit-bounds histogram layout mandated by the OTLP Trace + * Metrics Export RFC. + */ +final class OtlpHistogramBuckets { + private OtlpHistogramBuckets() {} + + private static final double NANOS_PER_SECOND = 1_000_000_000d; + + static final double[] BOUNDS_SECONDS = { + 0.002, 0.004, 0.006, 0.008, 0.01, 0.05, 0.1, 0.2, 0.4, 0.8, 1, 1.4, 2, 5, 10, 15 + }; + + static final List EXPLICIT_BOUNDS; + + static { + List bounds = new ArrayList<>(BOUNDS_SECONDS.length + 1); + for (double bound : BOUNDS_SECONDS) { + bounds.add(bound); + } + bounds.add(Double.POSITIVE_INFINITY); + EXPLICIT_BOUNDS = Collections.unmodifiableList(bounds); + } + + static int bucketIndex(double seconds) { + for (int i = 0; i < BOUNDS_SECONDS.length; i++) { + if (seconds <= BOUNDS_SECONDS[i]) { + return i; + } + } + return BOUNDS_SECONDS.length; // overflow + } + + /** + * Re-bins {@code histogram} (nanosecond-valued) into an {@link OtlpHistogramPoint} expressed in + * seconds with OTLP's fixed bucket layout. {@code count}, {@code min}, and {@code max} are taken + * directly from the sketch; {@code sum} is exact when the sketch tracks it ({@link + * HistogramWithSum}) and otherwise best-effort estimated from bin upper bounds. + */ + static OtlpHistogramPoint toHistogramPoint(Histogram histogram) { + long[] bucketCounts = new long[BOUNDS_SECONDS.length + 1]; + + List binBoundaries = histogram.getBinBoundaries(); + List binCounts = histogram.getBinCounts(); + double estimatedSumSeconds = 0d; + for (int i = 0; i < binBoundaries.size(); i++) { + double upperSeconds = binBoundaries.get(i) / NANOS_PER_SECOND; + long count = (long) binCounts.get(i).doubleValue(); + bucketCounts[bucketIndex(upperSeconds)] += count; + estimatedSumSeconds += upperSeconds * count; + } + + List counts = new ArrayList<>(bucketCounts.length); + for (long count : bucketCounts) { + counts.add((double) count); + } + + double sumSeconds = + histogram instanceof HistogramWithSum + ? ((HistogramWithSum) histogram).getSum() / NANOS_PER_SECOND + : estimatedSumSeconds; + + 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..e31371a6927 --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java @@ -0,0 +1,218 @@ +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 ConflatingMetricsAggregator} / {@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"; + + @Nullable private final OtlpSender sender; + + 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)); + } + + // visible for testing: lets tests inject a capturing sender to decode the emitted protobuf + OtlpStatsMetricWriter(@Nullable OtlpSender sender) { + this.sender = sender; + } + + @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.debug( + "Unsupported OTLP metrics protocol for trace metrics export: {}", + 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) { + 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); + OtlpHistogramPoint point = OtlpHistogramBuckets.toHistogramPoint(latencies); + metricBytes += recordDataPointMessage(buf, point, protobuf); + } + + private void writeDataPointAttributes(AggregateEntry entry, boolean error) { + // TODO(step 4): branch on isTraceOtelSemanticsEnabled() to add the datadog.* attribute set in + // default mode and to omit it in OTel-semantics mode. The OTel-semconv attributes below are + // emitted in both modes. + if (error) { + writeStringAttribute(STATUS_CODE, STATUS_CODE_ERROR); + } + writeStringAttribute(SPAN_NAME, entry.getResource()); + writeStringAttribute(SPAN_KIND, entry.getSpanKind()); + if (entry.getHttpMethod() != null) { + writeStringAttribute(HTTP_REQUEST_METHOD, entry.getHttpMethod()); + } + if (entry.getHttpStatusCode() != 0) { + writeLongAttribute(HTTP_RESPONSE_STATUS_CODE, entry.getHttpStatusCode()); + } + if (entry.getHttpEndpoint() != null) { + writeStringAttribute(HTTP_ROUTE, entry.getHttpEndpoint()); + } + if (entry.getGrpcStatusCode() != null) { + writeStringAttribute(RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode()); + } + } + + 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) { + scopedBytes += recordMetricMessage(buf, METRIC_DESCRIPTOR, metricBytes, protobuf); + } + 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/common/OtlpResourceProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java index 0e45aa22ab7..0a86668c234 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java @@ -47,6 +47,12 @@ static byte[] buildResourceMessage(Config config) { if (!version.isEmpty()) { writeResourceAttribute(buf, "service.version", version); } + if (config.isReportHostName()) { + String hostName = config.getHostName(); + if (hostName != null && !hostName.isEmpty()) { + writeResourceAttribute(buf, "host.name", hostName); + } + } writeResourceAttribute(buf, "telemetry.sdk.name", "datadog"); writeResourceAttribute(buf, "telemetry.sdk.version", TRACER_VERSION); writeResourceAttribute(buf, "telemetry.sdk.language", "java"); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java index 541f1750554..cb613137bcc 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java @@ -5,6 +5,7 @@ import static datadog.trace.api.config.GeneralConfig.SERVICE_NAME; import static datadog.trace.api.config.GeneralConfig.TAGS; import static datadog.trace.api.config.GeneralConfig.VERSION; +import static datadog.trace.api.config.TracerConfig.TRACE_REPORT_HOSTNAME; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -99,6 +100,16 @@ static Stream resourceMessageCases() { "service.name", "my-service", "region", "us-east", "team", "platform")), + // report-hostname disabled (default): no host.name written + Arguments.of( + "report-hostname disabled", + props(SERVICE_NAME, "my-service"), + attrs("service.name", "my-service")), + // report-hostname enabled: host.name written with the detected hostname + Arguments.of( + "report-hostname enabled", + props(SERVICE_NAME, "my-service", TRACE_REPORT_HOSTNAME, "true"), + attrs("service.name", "my-service", "host.name", Config.get().getHostName())), // all config values set together; telemetry.sdk.* keys in tags must be ignored Arguments.of( "service, env, version, and tags all set", From 095eec33b97a69f4def5912fdeba6b1c345c779d Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 17 Jun 2026 13:47:29 -0400 Subject: [PATCH 03/16] adding exact sums for ok/error --- .../common/metrics/OtlpHistogramBuckets.java | 15 ++++----------- .../common/metrics/OtlpStatsMetricWriter.java | 3 ++- 2 files changed, 6 insertions(+), 12 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java index 977bf74ead6..a07165484ed 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java @@ -1,7 +1,6 @@ package datadog.trace.common.metrics; import datadog.metrics.api.Histogram; -import datadog.metrics.api.HistogramWithSum; import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint; import java.util.ArrayList; import java.util.Collections; @@ -44,20 +43,18 @@ static int bucketIndex(double seconds) { /** * Re-bins {@code histogram} (nanosecond-valued) into an {@link OtlpHistogramPoint} expressed in * seconds with OTLP's fixed bucket layout. {@code count}, {@code min}, and {@code max} are taken - * directly from the sketch; {@code sum} is exact when the sketch tracks it ({@link - * HistogramWithSum}) and otherwise best-effort estimated from bin upper bounds. + * directly from the sketch; {@code sumNanos} is the exact duration sum tracked alongside the + * sketch by {@link AggregateEntry} (the DDSketch-derived sum would only be approximate). */ - static OtlpHistogramPoint toHistogramPoint(Histogram histogram) { + static OtlpHistogramPoint toHistogramPoint(Histogram histogram, long sumNanos) { long[] bucketCounts = new long[BOUNDS_SECONDS.length + 1]; List binBoundaries = histogram.getBinBoundaries(); List binCounts = histogram.getBinCounts(); - double estimatedSumSeconds = 0d; for (int i = 0; i < binBoundaries.size(); i++) { double upperSeconds = binBoundaries.get(i) / NANOS_PER_SECOND; long count = (long) binCounts.get(i).doubleValue(); bucketCounts[bucketIndex(upperSeconds)] += count; - estimatedSumSeconds += upperSeconds * count; } List counts = new ArrayList<>(bucketCounts.length); @@ -65,11 +62,7 @@ static OtlpHistogramPoint toHistogramPoint(Histogram histogram) { counts.add((double) count); } - double sumSeconds = - histogram instanceof HistogramWithSum - ? ((HistogramWithSum) histogram).getSum() / NANOS_PER_SECOND - : estimatedSumSeconds; - + 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; 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 index e31371a6927..a190c5748c2 100644 --- 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 @@ -141,7 +141,8 @@ private void addDataPoint(AggregateEntry entry, Histogram latencies, boolean err writeI64(buf, startNanos); writeTag(buf, DP_TIME_FIELD, I64_WIRE_TYPE); writeI64(buf, endNanos); - OtlpHistogramPoint point = OtlpHistogramBuckets.toHistogramPoint(latencies); + long sumNanos = error ? entry.getErrorDuration() : entry.getOkDuration(); + OtlpHistogramPoint point = OtlpHistogramBuckets.toHistogramPoint(latencies, sumNanos); metricBytes += recordDataPointMessage(buf, point, protobuf); } From 4c92994d0fa09c7a6559acdb1127d0b9a7ceb534 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 17 Jun 2026 13:56:54 -0400 Subject: [PATCH 04/16] remove host.name changes --- .../trace/core/otlp/common/OtlpResourceProto.java | 6 ------ .../trace/core/otlp/common/OtlpResourceProtoTest.java | 11 ----------- 2 files changed, 17 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java index 0a86668c234..0e45aa22ab7 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java @@ -47,12 +47,6 @@ static byte[] buildResourceMessage(Config config) { if (!version.isEmpty()) { writeResourceAttribute(buf, "service.version", version); } - if (config.isReportHostName()) { - String hostName = config.getHostName(); - if (hostName != null && !hostName.isEmpty()) { - writeResourceAttribute(buf, "host.name", hostName); - } - } writeResourceAttribute(buf, "telemetry.sdk.name", "datadog"); writeResourceAttribute(buf, "telemetry.sdk.version", TRACER_VERSION); writeResourceAttribute(buf, "telemetry.sdk.language", "java"); diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java index cb613137bcc..541f1750554 100644 --- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/common/OtlpResourceProtoTest.java @@ -5,7 +5,6 @@ import static datadog.trace.api.config.GeneralConfig.SERVICE_NAME; import static datadog.trace.api.config.GeneralConfig.TAGS; import static datadog.trace.api.config.GeneralConfig.VERSION; -import static datadog.trace.api.config.TracerConfig.TRACE_REPORT_HOSTNAME; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -100,16 +99,6 @@ static Stream resourceMessageCases() { "service.name", "my-service", "region", "us-east", "team", "platform")), - // report-hostname disabled (default): no host.name written - Arguments.of( - "report-hostname disabled", - props(SERVICE_NAME, "my-service"), - attrs("service.name", "my-service")), - // report-hostname enabled: host.name written with the detected hostname - Arguments.of( - "report-hostname enabled", - props(SERVICE_NAME, "my-service", TRACE_REPORT_HOSTNAME, "true"), - attrs("service.name", "my-service", "host.name", Config.get().getHostName())), // all config values set together; telemetry.sdk.* keys in tags must be ignored Arguments.of( "service, env, version, and tags all set", From da4d67d1d4244babd2669d687cae5b363e2c0f3b Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 17 Jun 2026 16:27:52 -0400 Subject: [PATCH 05/16] init tests --- .../metrics/OtlpHistogramBucketsTest.java | 153 +++++ .../metrics/OtlpStatsMetricWriterTest.java | 534 ++++++++++++++++++ 2 files changed, 687 insertions(+) create mode 100644 dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java create mode 100644 dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java new file mode 100644 index 00000000000..ad0de52e0ec --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java @@ -0,0 +1,153 @@ +package datadog.trace.common.metrics; + +import static datadog.trace.common.metrics.OtlpHistogramBuckets.BOUNDS_SECONDS; +import static datadog.trace.common.metrics.OtlpHistogramBuckets.EXPLICIT_BOUNDS; +import static datadog.trace.common.metrics.OtlpHistogramBuckets.bucketIndex; +import static datadog.trace.common.metrics.OtlpHistogramBuckets.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 OtlpHistogramBuckets}, the helper that re-bins a nanosecond-valued DDSketch + * onto the fixed OTLP explicit-bounds histogram layout (in seconds). + * + *

Both methods under test ({@code bucketIndex}, {@code toHistogramPoint}) are package-private, + * so this test lives in {@code datadog.trace.common.metrics}. + */ +class OtlpHistogramBucketsTest { + + 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(); + // Samples spanning the layout, including a 1ns duration — the upstream-clamped minimum + // (DDSpan#finishAndAddToTrace does Math.max(1, durationNano)). 1ns is far below the smallest + // bound (0.002s) and must land in bucket 0, not be dropped. 1ms also falls in bucket 0, so the + // two sub-2ms samples share it; 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); + + // The two sub-2ms samples (1ns + 1ms) both land in bucket 0; nothing is lost: bucket counts + // sum to the total count. The count==Σbuckets invariant guards against a future histogram + // config that parks tiny values in a zero/negative store (excluded from getBinCounts). + 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..c47973c257f --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java @@ -0,0 +1,534 @@ +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 org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + +/** + * 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}. + * + *

Lives in {@code datadog.trace.common.metrics} to reach the package-private testing constructor + * {@code OtlpStatsMetricWriter(OtlpSender)} and the {@code AggregateEntryTestUtils} factory. + * + *

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; + + @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 ────────────────────────────────────────────────────── + + /** Build an entry and record {@code hits} ok durations of {@code durationNanos} each. */ + private static AggregateEntry okEntry(long durationNanos, int hits) { + AggregateEntry e = + AggregateEntryTestUtils.of( + "GET /users", + "web", + "servlet.request", + null, + "web", + 0, + false, + true, + "server", + null, + 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. */ + private static final class DataPoint { + long start; + long end; + long count; + double sum; + double min; + double max; + final List bucketCounts = new ArrayList<>(); + final List bounds = new ArrayList<>(); + 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 5: + p.sum = dp.readDouble(); + break; + case 6: + p.bucketCounts.add(dp.readFixed64()); + break; + case 7: + p.bounds.add(dp.readDouble()); + break; + case 9: // attributes (KeyValue) + readKeyValue(dp.readBytes().newCodedInput(), p.attributes); + break; + case 11: + p.min = dp.readDouble(); + break; + case 12: + p.max = dp.readDouble(); + break; + default: + 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; + } + + // ── test cases ────────────────────────────────────────────────────────── + + @Test + void okOnlyEntryProducesExactlyOneDataPoint() throws IOException { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender); + + long start = SECONDS.toNanos(1_700_000_000L); + long duration = SECONDS.toNanos(10); + writer.startBucket(1, start, duration); + writer.add(okEntry(SECONDS.toNanos(1), 3)); + writer.finishBucket(); + + assertEquals(1, sender.sendCount, "exactly one payload sent"); + DecodedMetric metric = decode(sender.lastPayload); + + 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(start, dp.start, "start_time_unix_nano == startBucket start"); + assertEquals(start + duration, dp.end, "time_unix_nano == start + duration"); + assertEquals(3L, dp.count); + assertEquals(3L, sumBuckets(dp), "bucket counts sum to total count"); + assertEquals(17, dp.bucketCounts.size(), "17 OTLP buckets"); + assertFalse(dp.attributes.containsKey("status.code"), "ok point carries no status.code"); + } + + @Test + void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender); + + AggregateEntry e = + AggregateEntryTestUtils.of( + "GET /users", + "web", + "servlet.request", + null, + "web", + 0, + false, + true, + "server", + null, + null, + null, + null); + e.recordOneDuration(SECONDS.toNanos(1)); // ok + e.recordOneDuration(SECONDS.toNanos(2)); // ok + e.recordOneDuration(SECONDS.toNanos(3) | AggregateEntry.ERROR_TAG); // error + + long start = SECONDS.toNanos(1_700_000_000L); + long duration = SECONDS.toNanos(10); + writer.startBucket(1, start, duration); + writer.add(e); + writer.finishBucket(); + + DecodedMetric metric = decode(sender.lastPayload); + 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); + assertEquals(okCount, sumBuckets(okPoint)); + assertEquals(errorCount, sumBuckets(errorPoint)); + } + + @Test + void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender); + + AggregateEntry e = + AggregateEntryTestUtils.of( + "GET /users/{id}", + "web", + "servlet.request", + null, + "web", + 200, + false, + true, + "server", + null, + "GET", + "/users/{id}", + "0"); + e.recordOneDuration(SECONDS.toNanos(1)); + + long start = SECONDS.toNanos(1_700_000_000L); + writer.startBucket(1, start, SECONDS.toNanos(10)); + writer.add(e); + writer.finishBucket(); + + DecodedMetric metric = decode(sender.lastPayload); + 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 + CapturingSender sender2 = new CapturingSender(); + OtlpStatsMetricWriter writer2 = new OtlpStatsMetricWriter(sender2); + writer2.startBucket(1, start, SECONDS.toNanos(10)); + writer2.add(okEntry(SECONDS.toNanos(1), 1)); + writer2.finishBucket(); + Map bareAttrs = decode(sender2.lastPayload).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); + + writer.startBucket(0, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); + 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((OtlpSender) null); + writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); + 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); + } + } + + // ── step 4: attribute modes ────────────────────────────────────────────── + + @Test + void defaultModeCarriesDatadogAttributes() throws IOException { + CapturingSender sender = new CapturingSender(); + // otelSemanticsMode = false → datadog.* should be present + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false); + writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); + // use an entry where all hits are top-level: OR in TOP_LEVEL_TAG + AggregateEntry e = + AggregateEntryTestUtils.of( + "servlet.request", + "web", + "servlet.request", + null, + "web", + 0, + false, + true, + "server", + null, + null, + null, + null); + e.recordOneDuration(SECONDS.toNanos(1) | AggregateEntry.TOP_LEVEL_TAG); + writer.add(e); + writer.finishBucket(); + + Map attrs = decode(sender.lastPayload).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"); + assertFalse(attrs.containsKey("datadog.origin"), "non-synthetic entry has no datadog.origin"); + } + + @Test + void defaultModeCarriesSyntheticsOrigin() throws IOException { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false); + writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); + AggregateEntry e = + AggregateEntryTestUtils.of( + "servlet.request", + "web", + "servlet.request", + null, + "web", + 0, + true, // synthetic=true + true, + "server", + null, + null, + null, + null); + e.recordOneDuration(SECONDS.toNanos(1)); + writer.add(e); + writer.finishBucket(); + + Map attrs = decode(sender.lastPayload).dataPoints.get(0).attributes; + assertEquals( + "synthetics", attrs.get("datadog.origin"), "synthetic entry carries datadog.origin"); + } + + @Test + void otelSemanticsModeOmitsDatadogAttributes() throws IOException { + CapturingSender sender = new CapturingSender(); + // otelSemanticsMode = true → datadog.* must be absent + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, true); + writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); + writer.add(okEntry(SECONDS.toNanos(1), 1)); + writer.finishBucket(); + + Map attrs = decode(sender.lastPayload).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"); + } + + private static long sumBuckets(DataPoint dp) { + long total = 0; + for (long c : dp.bucketCounts) { + total += c; + } + return total; + } +} From 5ef4652b85575828368735b3451808428e619fb4 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Thu, 18 Jun 2026 13:11:02 -0400 Subject: [PATCH 06/16] adding datadog-specific datapoint attribtues --- .../common/metrics/OtlpStatsMetricWriter.java | 27 ++++++++++++++++--- .../metrics/OtlpStatsMetricWriterTest.java | 17 ++++++++++++ 2 files changed, 40 insertions(+), 4 deletions(-) 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 index a190c5748c2..7103c3d7092 100644 --- 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 @@ -63,8 +63,13 @@ public final class OtlpStatsMetricWriter implements MetricWriter { 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"; @Nullable private final OtlpSender sender; + private final boolean otelSemanticsMode; private final GrowableBuffer buf = new GrowableBuffer(512); private final OtlpProtoBuffer protobuf = new OtlpProtoBuffer(8192); @@ -77,12 +82,18 @@ public final class OtlpStatsMetricWriter implements MetricWriter { private int metricBytes; public OtlpStatsMetricWriter(Config config) { - this(createSender(config)); + this(createSender(config), config.isTraceOtelSemanticsEnabled()); } // visible for testing: lets tests inject a capturing sender to decode the emitted protobuf OtlpStatsMetricWriter(@Nullable OtlpSender sender) { + this(sender, false); + } + + // visible for testing: lets tests inject a capturing sender and control the semantics mode + OtlpStatsMetricWriter(@Nullable OtlpSender sender, boolean otelSemanticsMode) { this.sender = sender; + this.otelSemanticsMode = otelSemanticsMode; } @Nullable @@ -147,12 +158,10 @@ private void addDataPoint(AggregateEntry entry, Histogram latencies, boolean err } private void writeDataPointAttributes(AggregateEntry entry, boolean error) { - // TODO(step 4): branch on isTraceOtelSemanticsEnabled() to add the datadog.* attribute set in - // default mode and to omit it in OTel-semantics mode. The OTel-semconv attributes below are - // emitted in both modes. 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.getHttpMethod() != null) { @@ -167,6 +176,16 @@ private void writeDataPointAttributes(AggregateEntry entry, boolean error) { if (entry.getGrpcStatusCode() != null) { 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); + // Emit the full origin (synthetics, synthetics-browser, rum, ciapp-test, lambda, ...) when + // present; writeStringAttribute no-ops on a null value. + writeStringAttribute(DATADOG_ORIGIN, entry.getOrigin()); + } } private void writeStringAttribute(String key, @Nullable UTF8BytesString value) { 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 index c47973c257f..cffa781cc15 100644 --- 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 @@ -524,6 +524,23 @@ void otelSemanticsModeOmitsDatadogAttributes() throws IOException { assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode"); } + @Test + void defaultModePassesFullOriginThrough() throws IOException { + // A non-synthetics origin must be emitted verbatim, not collapsed to a synthetics flag. + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false); + writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); + AggregateEntry e = + AggregateEntryTestUtils.ofOrigin( + "servlet.request", "web", "servlet.request", "web", "server", "ciapp-test"); + e.recordOneDuration(SECONDS.toNanos(1)); + writer.add(e); + writer.finishBucket(); + + Map attrs = decode(sender.lastPayload).dataPoints.get(0).attributes; + assertEquals("ciapp-test", attrs.get("datadog.origin"), "full origin emitted verbatim"); + } + private static long sumBuckets(DataPoint dp) { long total = 0; for (long c : dp.bucketCounts) { From a1a428c973429a49b91fab596f80520c5cf5f13a Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Tue, 23 Jun 2026 09:38:49 -0400 Subject: [PATCH 07/16] more specific logging --- .../datadog/trace/common/metrics/OtlpStatsMetricWriter.java | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) 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 index 7103c3d7092..28ec2c3bed3 100644 --- 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 @@ -116,8 +116,9 @@ private static OtlpSender createSender(Config config) { config.getOtlpMetricsCompression()); default: // HTTP_JSON has no protobuf-free encoder yet; JSON transport is deferred per the plan. - log.debug( - "Unsupported OTLP metrics protocol for trace metrics export: {}", + 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; } From 72c12bb847d2b0e585e1d243589777fed79324a7 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 24 Jun 2026 14:03:17 -0400 Subject: [PATCH 08/16] refactor origin usage in test helpers and add origin specific testS remove irrelevant test following rebase --- .../metrics/OtlpStatsMetricWriterTest.java | 18 ------------------ 1 file changed, 18 deletions(-) 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 index cffa781cc15..67c5fdae106 100644 --- 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 @@ -523,24 +523,6 @@ void otelSemanticsModeOmitsDatadogAttributes() throws IOException { // OTel-semconv attrs must still be present assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode"); } - - @Test - void defaultModePassesFullOriginThrough() throws IOException { - // A non-synthetics origin must be emitted verbatim, not collapsed to a synthetics flag. - CapturingSender sender = new CapturingSender(); - OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false); - writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10)); - AggregateEntry e = - AggregateEntryTestUtils.ofOrigin( - "servlet.request", "web", "servlet.request", "web", "server", "ciapp-test"); - e.recordOneDuration(SECONDS.toNanos(1)); - writer.add(e); - writer.finishBucket(); - - Map attrs = decode(sender.lastPayload).dataPoints.get(0).attributes; - assertEquals("ciapp-test", attrs.get("datadog.origin"), "full origin emitted verbatim"); - } - private static long sumBuckets(DataPoint dp) { long total = 0; for (long c : dp.bucketCounts) { From 966296715e724c3fce0036f1dcf16882d6c3c5f0 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Thu, 25 Jun 2026 22:59:23 -0400 Subject: [PATCH 09/16] rename histogram class and refactor tests --- ...ts.java => OtlpStatsHistogramBuckets.java} | 8 +- .../common/metrics/OtlpStatsMetricWriter.java | 24 +- ...ava => OtlpStatsHistogramBucketsTest.java} | 26 +- .../metrics/OtlpStatsMetricWriterTest.java | 269 ++++++------------ 4 files changed, 112 insertions(+), 215 deletions(-) rename dd-trace-core/src/main/java/datadog/trace/common/metrics/{OtlpHistogramBuckets.java => OtlpStatsHistogramBuckets.java} (86%) rename dd-trace-core/src/test/java/datadog/trace/common/metrics/{OtlpHistogramBucketsTest.java => OtlpStatsHistogramBucketsTest.java} (80%) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsHistogramBuckets.java similarity index 86% rename from dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java rename to dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsHistogramBuckets.java index a07165484ed..372cac73bae 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpHistogramBuckets.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsHistogramBuckets.java @@ -11,8 +11,8 @@ * nanoseconds) onto the fixed explicit-bounds histogram layout mandated by the OTLP Trace * Metrics Export RFC. */ -final class OtlpHistogramBuckets { - private OtlpHistogramBuckets() {} +final class OtlpStatsHistogramBuckets { + private OtlpStatsHistogramBuckets() {} private static final double NANOS_PER_SECOND = 1_000_000_000d; @@ -42,9 +42,7 @@ static int bucketIndex(double seconds) { /** * Re-bins {@code histogram} (nanosecond-valued) into an {@link OtlpHistogramPoint} expressed in - * seconds with OTLP's fixed bucket layout. {@code count}, {@code min}, and {@code max} are taken - * directly from the sketch; {@code sumNanos} is the exact duration sum tracked alongside the - * sketch by {@link AggregateEntry} (the DDSketch-derived sum would only be approximate). + * seconds with OTLP's fixed bucket layout. */ static OtlpHistogramPoint toHistogramPoint(Histogram histogram, long sumNanos) { long[] bucketCounts = new long[BOUNDS_SECONDS.length + 1]; 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 index 28ec2c3bed3..b199bf745fe 100644 --- 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 @@ -67,10 +67,12 @@ public final class OtlpStatsMetricWriter implements MetricWriter { 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); @@ -85,12 +87,8 @@ public OtlpStatsMetricWriter(Config config) { this(createSender(config), config.isTraceOtelSemanticsEnabled()); } - // visible for testing: lets tests inject a capturing sender to decode the emitted protobuf - OtlpStatsMetricWriter(@Nullable OtlpSender sender) { - this(sender, false); - } - - // visible for testing: lets tests inject a capturing sender and control the semantics mode + // 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; @@ -154,7 +152,7 @@ private void addDataPoint(AggregateEntry entry, Histogram latencies, boolean err writeTag(buf, DP_TIME_FIELD, I64_WIRE_TYPE); writeI64(buf, endNanos); long sumNanos = error ? entry.getErrorDuration() : entry.getOkDuration(); - OtlpHistogramPoint point = OtlpHistogramBuckets.toHistogramPoint(latencies, sumNanos); + OtlpHistogramPoint point = OtlpStatsHistogramBuckets.toHistogramPoint(latencies, sumNanos); metricBytes += recordDataPointMessage(buf, point, protobuf); } @@ -165,16 +163,16 @@ private void writeDataPointAttributes(AggregateEntry entry, boolean error) { // OTel semconv attrs are emitted in both modes writeStringAttribute(SPAN_NAME, entry.getResource()); writeStringAttribute(SPAN_KIND, entry.getSpanKind()); - if (entry.getHttpMethod() != null) { + if (entry.hasHttpMethod()) { writeStringAttribute(HTTP_REQUEST_METHOD, entry.getHttpMethod()); } if (entry.getHttpStatusCode() != 0) { writeLongAttribute(HTTP_RESPONSE_STATUS_CODE, entry.getHttpStatusCode()); } - if (entry.getHttpEndpoint() != null) { + if (entry.hasHttpEndpoint()) { writeStringAttribute(HTTP_ROUTE, entry.getHttpEndpoint()); } - if (entry.getGrpcStatusCode() != null) { + if (entry.hasGrpcStatusCode()) { writeStringAttribute(RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode()); } // Default (Datadog) mode: emit datadog.* per-point attributes @@ -183,9 +181,9 @@ private void writeDataPointAttributes(AggregateEntry entry, boolean error) { writeStringAttribute(DATADOG_SPAN_TYPE, entry.getType()); writeLongAttribute( DATADOG_SPAN_TOP_LEVEL, entry.getTopLevelCount() == entry.getHitCount() ? 1L : 0L); - // Emit the full origin (synthetics, synthetics-browser, rum, ciapp-test, lambda, ...) when - // present; writeStringAttribute no-ops on a null value. - writeStringAttribute(DATADOG_ORIGIN, entry.getOrigin()); + if (entry.isSynthetics()) { + writeStringAttribute(DATADOG_ORIGIN, SYNTHETICS_ORIGIN); + } } } diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java similarity index 80% rename from dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java rename to dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java index ad0de52e0ec..d37cde2d92a 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpHistogramBucketsTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java @@ -1,9 +1,9 @@ package datadog.trace.common.metrics; -import static datadog.trace.common.metrics.OtlpHistogramBuckets.BOUNDS_SECONDS; -import static datadog.trace.common.metrics.OtlpHistogramBuckets.EXPLICIT_BOUNDS; -import static datadog.trace.common.metrics.OtlpHistogramBuckets.bucketIndex; -import static datadog.trace.common.metrics.OtlpHistogramBuckets.toHistogramPoint; +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; @@ -20,13 +20,10 @@ import org.junit.jupiter.params.provider.MethodSource; /** - * Unit tests for {@link OtlpHistogramBuckets}, the helper that re-bins a nanosecond-valued DDSketch - * onto the fixed OTLP explicit-bounds histogram layout (in seconds). - * - *

Both methods under test ({@code bucketIndex}, {@code toHistogramPoint}) are package-private, - * so this test lives in {@code datadog.trace.common.metrics}. + * 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 OtlpHistogramBucketsTest { +class OtlpStatsHistogramBucketsTest { private static final double NANOS_PER_SECOND = 1_000_000_000d; @@ -85,10 +82,8 @@ void explicitBoundsHas17EntriesEndingInInfinity() { @Test void toHistogramPointSummary() { Histogram h = Histogram.newHistogram(); - // Samples spanning the layout, including a 1ns duration — the upstream-clamped minimum - // (DDSpan#finishAndAddToTrace does Math.max(1, durationNano)). 1ns is far below the smallest - // bound (0.002s) and must land in bucket 0, not be dropped. 1ms also falls in bucket 0, so the - // two sub-2ms samples share it; 100ms → bucket 6, 3s → bucket 13. + // 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), @@ -122,9 +117,6 @@ void toHistogramPointSummary() { // sum equals the sumNanos ARGUMENT converted to seconds, not the sketch sum. assertEquals(42.0, point.sum, EPS); - // The two sub-2ms samples (1ns + 1ms) both land in bucket 0; nothing is lost: bucket counts - // sum to the total count. The count==Σbuckets invariant guards against a future histogram - // config that parks tiny values in a zero/negative store (excluded from getBinCounts). long bucketTotal = 0; for (int i = 0; i < point.bucketCounts.size(); i++) { long c = (long) point.bucketCounts.get(i).doubleValue(); 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 index 67c5fdae106..92088f510cf 100644 --- 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 @@ -19,8 +19,11 @@ 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 @@ -28,9 +31,6 @@ * protobuf {@code ExportMetricsServiceRequest} ({@code MetricsData}) using protobuf's {@link * CodedInputStream}, reusing the decode idioms from {@code OtlpMetricsProtoTest}. * - *

Lives in {@code datadog.trace.common.metrics} to reach the package-private testing constructor - * {@code OtlpStatsMetricWriter(OtlpSender)} and the {@code AggregateEntryTestUtils} factory. - * *

Wire layout (OTLP metrics proto): * * 

@@ -47,6 +47,8 @@
 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() {
@@ -74,23 +76,32 @@ 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 =
-        AggregateEntryTestUtils.of(
-            "GET /users",
-            "web",
-            "servlet.request",
-            null,
-            "web",
-            0,
-            false,
-            true,
-            "server",
-            null,
-            null,
-            null,
-            null);
+    AggregateEntry e = entry("GET /users", false, 0, null, null, null);
     for (int i = 0; i < hits; i++) {
       e.recordOneDuration(durationNanos);
     }
@@ -99,16 +110,15 @@ private static AggregateEntry okEntry(long durationNanos, int hits) {
 
   // ── decode helpers (adapted from OtlpMetricsProtoTest) ──────────────────────
 
-  /** A decoded histogram data point. */
+  /**
+   * 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;
-    double sum;
-    double min;
-    double max;
-    final List bucketCounts = new ArrayList<>();
-    final List bounds = new ArrayList<>();
     final Map attributes = new HashMap<>();
   }
 
@@ -206,25 +216,10 @@ private static DataPoint parseDataPoint(CodedInputStream dp) throws IOException
         case 4:
           p.count = dp.readFixed64();
           break;
-        case 5:
-          p.sum = dp.readDouble();
-          break;
-        case 6:
-          p.bucketCounts.add(dp.readFixed64());
-          break;
-        case 7:
-          p.bounds.add(dp.readDouble());
-          break;
         case 9: // attributes (KeyValue)
           readKeyValue(dp.readBytes().newCodedInput(), p.attributes);
           break;
-        case 11:
-          p.min = dp.readDouble();
-          break;
-        case 12:
-          p.max = dp.readDouble();
-          break;
-        default:
+        default: // sum, bucket_counts, explicit_bounds, min, max — not asserted here
           dp.skipField(tag);
       }
     }
@@ -275,21 +270,29 @@ private static Object readAnyValue(CodedInputStream any) throws IOException {
     return value;
   }
 
-  // ── test cases ──────────────────────────────────────────────────────────
+  // ── writer driver ─────────────────────────────────────────────────────────
 
-  @Test
-  void okOnlyEntryProducesExactlyOneDataPoint() throws IOException {
+  /**
+   * 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);
-
-    long start = SECONDS.toNanos(1_700_000_000L);
-    long duration = SECONDS.toNanos(10);
-    writer.startBucket(1, start, duration);
-    writer.add(okEntry(SECONDS.toNanos(1), 3));
+    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");
-    DecodedMetric metric = decode(sender.lastPayload);
+    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);
@@ -297,45 +300,20 @@ void okOnlyEntryProducesExactlyOneDataPoint() throws IOException {
     assertEquals(1, metric.dataPoints.size(), "ok-only → one data point");
 
     DataPoint dp = metric.dataPoints.get(0);
-    assertEquals(start, dp.start, "start_time_unix_nano == startBucket start");
-    assertEquals(start + duration, dp.end, "time_unix_nano == start + duration");
+    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);
-    assertEquals(3L, sumBuckets(dp), "bucket counts sum to total count");
-    assertEquals(17, dp.bucketCounts.size(), "17 OTLP buckets");
     assertFalse(dp.attributes.containsKey("status.code"), "ok point carries no status.code");
   }
 
   @Test
   void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
-    CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
-
-    AggregateEntry e =
-        AggregateEntryTestUtils.of(
-            "GET /users",
-            "web",
-            "servlet.request",
-            null,
-            "web",
-            0,
-            false,
-            true,
-            "server",
-            null,
-            null,
-            null,
-            null);
+    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
 
-    long start = SECONDS.toNanos(1_700_000_000L);
-    long duration = SECONDS.toNanos(10);
-    writer.startBucket(1, start, duration);
-    writer.add(e);
-    writer.finishBucket();
-
-    DecodedMetric metric = decode(sender.lastPayload);
+    DecodedMetric metric = writeAndDecode(false, e);
     assertEquals(2, metric.dataPoints.size(), "ok+error → two data points");
 
     long okCount = 0;
@@ -355,38 +333,14 @@ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
     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);
-    assertEquals(okCount, sumBuckets(okPoint));
-    assertEquals(errorCount, sumBuckets(errorPoint));
   }
 
   @Test
   void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
-    CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
-
-    AggregateEntry e =
-        AggregateEntryTestUtils.of(
-            "GET /users/{id}",
-            "web",
-            "servlet.request",
-            null,
-            "web",
-            200,
-            false,
-            true,
-            "server",
-            null,
-            "GET",
-            "/users/{id}",
-            "0");
+    AggregateEntry e = entry("GET /users/{id}", false, 200, "GET", "/users/{id}", "0");
     e.recordOneDuration(SECONDS.toNanos(1));
 
-    long start = SECONDS.toNanos(1_700_000_000L);
-    writer.startBucket(1, start, SECONDS.toNanos(10));
-    writer.add(e);
-    writer.finishBucket();
-
-    DecodedMetric metric = decode(sender.lastPayload);
+    DecodedMetric metric = writeAndDecode(false, e);
     assertEquals(1, metric.dataPoints.size());
     Map attrs = metric.dataPoints.get(0).attributes;
 
@@ -396,12 +350,8 @@ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
     assertEquals("0", attrs.get("rpc.response.status_code"));
 
     // a bare entry has none of these
-    CapturingSender sender2 = new CapturingSender();
-    OtlpStatsMetricWriter writer2 = new OtlpStatsMetricWriter(sender2);
-    writer2.startBucket(1, start, SECONDS.toNanos(10));
-    writer2.add(okEntry(SECONDS.toNanos(1), 1));
-    writer2.finishBucket();
-    Map bareAttrs = decode(sender2.lastPayload).dataPoints.get(0).attributes;
+    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"));
@@ -411,9 +361,9 @@ void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
   @Test
   void emptyBucketSendsNothing() {
     CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender);
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false);
 
-    writer.startBucket(0, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
+    writer.startBucket(0, BUCKET_START, BUCKET_DURATION);
     writer.finishBucket(); // no add()
 
     assertEquals(0, sender.sendCount, "empty bucket must not invoke send");
@@ -423,8 +373,8 @@ void emptyBucketSendsNothing() {
   @Test
   void nullSenderDoesNotThrowOnNonEmptyBucket() {
     // mirrors the HTTP_JSON path where createSender(config) returns null.
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter((OtlpSender) null);
-    writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(null, false);
+    writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
     writer.add(okEntry(SECONDS.toNanos(1), 2));
     try {
       writer.finishBucket();
@@ -433,35 +383,13 @@ void nullSenderDoesNotThrowOnNonEmptyBucket() {
     }
   }
 
-  // ── step 4: attribute modes ──────────────────────────────────────────────
-
   @Test
   void defaultModeCarriesDatadogAttributes() throws IOException {
-    CapturingSender sender = new CapturingSender();
-    // otelSemanticsMode = false → datadog.* should be present
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false);
-    writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
     // use an entry where all hits are top-level: OR in TOP_LEVEL_TAG
-    AggregateEntry e =
-        AggregateEntryTestUtils.of(
-            "servlet.request",
-            "web",
-            "servlet.request",
-            null,
-            "web",
-            0,
-            false,
-            true,
-            "server",
-            null,
-            null,
-            null,
-            null);
+    AggregateEntry e = entry("servlet.request", false, 0, null, null, null);
     e.recordOneDuration(SECONDS.toNanos(1) | AggregateEntry.TOP_LEVEL_TAG);
-    writer.add(e);
-    writer.finishBucket();
 
-    Map attrs = decode(sender.lastPayload).dataPoints.get(0).attributes;
+    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");
@@ -470,48 +398,36 @@ void defaultModeCarriesDatadogAttributes() throws IOException {
     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");
-    assertFalse(attrs.containsKey("datadog.origin"), "non-synthetic entry has no datadog.origin");
+    // datadog.origin presence/absence is covered by defaultModeEmitsSyntheticOrigin
   }
 
-  @Test
-  void defaultModeCarriesSyntheticsOrigin() throws IOException {
-    CapturingSender sender = new CapturingSender();
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false);
-    writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
-    AggregateEntry e =
-        AggregateEntryTestUtils.of(
-            "servlet.request",
-            "web",
-            "servlet.request",
-            null,
-            "web",
-            0,
-            true, // synthetic=true
-            true,
-            "server",
-            null,
-            null,
-            null,
-            null);
+  /**
+   * 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));
-    writer.add(e);
-    writer.finishBucket();
 
-    Map attrs = decode(sender.lastPayload).dataPoints.get(0).attributes;
-    assertEquals(
-        "synthetics", attrs.get("datadog.origin"), "synthetic entry carries datadog.origin");
+    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 {
-    CapturingSender sender = new CapturingSender();
     // otelSemanticsMode = true → datadog.* must be absent
-    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, true);
-    writer.startBucket(1, SECONDS.toNanos(1_700_000_000L), SECONDS.toNanos(10));
-    writer.add(okEntry(SECONDS.toNanos(1), 1));
-    writer.finishBucket();
-
-    Map attrs = decode(sender.lastPayload).dataPoints.get(0).attributes;
+    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");
@@ -523,11 +439,4 @@ void otelSemanticsModeOmitsDatadogAttributes() throws IOException {
     // OTel-semconv attrs must still be present
     assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode");
   }
-  private static long sumBuckets(DataPoint dp) {
-    long total = 0;
-    for (long c : dp.bucketCounts) {
-      total += c;
-    }
-    return total;
-  }
 }

From aa351f0c04b872ae462e48cc19f3860645878e55 Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Sat, 27 Jun 2026 09:11:21 -0400
Subject: [PATCH 10/16] force temporality delta on tracemetrics histogram

---
 .../common/metrics/OtlpStatsMetricWriter.java    |  4 +++-
 .../core/otlp/metrics/OtlpMetricsProto.java      | 16 +++++++++++++++-
 2 files changed, 18 insertions(+), 2 deletions(-)

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
index b199bf745fe..517077ed90b 100644
--- 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
@@ -207,7 +207,9 @@ private void writeLongAttribute(String key, long value) {
   public void finishBucket() {
     try {
       if (metricBytes > 0) {
-        scopedBytes += recordMetricMessage(buf, METRIC_DESCRIPTOR, metricBytes, protobuf);
+        // 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);
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());

From 9cf8a2566a0684ae01c4ed14664c8424550fd2dd Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Sat, 27 Jun 2026 09:27:02 -0400
Subject: [PATCH 11/16] gating error latencies on empty as well

---
 .../common/metrics/OtlpStatsMetricWriter.java |  2 +-
 .../metrics/OtlpStatsMetricWriterTest.java    | 35 +++++++++++++++++++
 2 files changed, 36 insertions(+), 1 deletion(-)

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
index 517077ed90b..62518b80785 100644
--- 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
@@ -140,7 +140,7 @@ public void add(AggregateEntry entry) {
     }
 
     Histogram errorLatencies = entry.getErrorLatencies();
-    if (errorLatencies != null) {
+    if (errorLatencies != null && !errorLatencies.isEmpty()) {
       addDataPoint(entry, errorLatencies, true);
     }
   }
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
index 92088f510cf..c825ed61d6b 100644
--- 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
@@ -335,6 +335,41 @@ void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException {
     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", null, 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");

From f04268a12c7de44430f09489113061c458855c47 Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Tue, 30 Jun 2026 15:14:29 -0400
Subject: [PATCH 12/16] update metricintegrationtest

---
 .../datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

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
index c825ed61d6b..31ed707cabd 100644
--- 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
@@ -341,7 +341,7 @@ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOExcept
     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", null, 0, null, null, null);
+    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
 

From 7b264b95e04eca33fd27584fcd43e61008f4de00 Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Thu, 2 Jul 2026 14:56:21 -0500
Subject: [PATCH 13/16] cleanup

---
 .../datadog/trace/common/metrics/OtlpStatsMetricWriter.java | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

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
index 62518b80785..98256d61d4f 100644
--- 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
@@ -34,9 +34,9 @@
  * (unit {@code s}).
  *
  * 
This is the parallel-to-{@link SerializingMetricWriter} OTLP export path. It hangs off the
- * same in-memory aggregation ({@link ConflatingMetricsAggregator} / {@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).
+ * 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}
  */

From 0a444edfc5742c9f0204ae76b4717d31a5eb5dba Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Thu, 9 Jul 2026 14:21:42 -0400
Subject: [PATCH 14/16] make AggregateEntry public and move trace metrics files
 to datadog.trace.core.otlp.metrics

---
 .../trace/common/metrics/AggregateEntry.java  | 52 +++++++++----------
 .../metrics/OtlpMetricsSenderFactory.java     | 43 +++++++++++++++
 .../core/otlp/metrics/OtlpMetricsService.java | 33 +++---------
 .../metrics/OtlpStatsHistogramBuckets.java    |  2 +-
 .../otlp}/metrics/OtlpStatsMetricWriter.java  | 48 +++++------------
 .../metrics/AggregateEntryTestUtils.java      | 24 +++++++++
 .../OtlpStatsHistogramBucketsTest.java        | 10 ++--
 .../metrics/OtlpStatsMetricWriterTest.java    | 26 +++++-----
 8 files changed, 133 insertions(+), 105 deletions(-)
 create mode 100644 dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsSenderFactory.java
 rename dd-trace-core/src/main/java/datadog/trace/{common => core/otlp}/metrics/OtlpStatsHistogramBuckets.java (98%)
 rename dd-trace-core/src/main/java/datadog/trace/{common => core/otlp}/metrics/OtlpStatsMetricWriter.java (84%)
 rename dd-trace-core/src/test/java/datadog/trace/{common => core/otlp}/metrics/OtlpStatsHistogramBucketsTest.java (93%)
 rename dd-trace-core/src/test/java/datadog/trace/{common => core/otlp}/metrics/OtlpStatsMetricWriterTest.java (95%)

diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java
index b18fce272a4..32c0ad7a056 100644
--- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java
+++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/AggregateEntry.java
@@ -21,7 +21,7 @@
  * static); tests using {@link PeerTagSchema} must also call {@code PeerTagSchema#resetHandlers} on
  * the schema instance.
  */
-final class AggregateEntry extends Hashtable.Entry {
+public final class AggregateEntry extends Hashtable.Entry {
 
   static final long ERROR_TAG = 0x8000000000000000L;
   static final long TOP_LEVEL_TAG = 0x4000000000000000L;
@@ -165,31 +165,31 @@ AggregateEntry recordOneDuration(long tagAndDuration) {
     return this;
   }
 
-  int getErrorCount() {
+  public int getErrorCount() {
     return errorCount;
   }
 
-  int getHitCount() {
+  public int getHitCount() {
     return hitCount;
   }
 
-  int getTopLevelCount() {
+  public int getTopLevelCount() {
     return topLevelCount;
   }
 
-  long getDuration() {
+  public long getDuration() {
     return okDuration + errorDuration;
   }
 
-  long getOkDuration() {
+  public long getOkDuration() {
     return okDuration;
   }
 
-  long getErrorDuration() {
+  public long getErrorDuration() {
     return errorDuration;
   }
 
-  Histogram getOkLatencies() {
+  public Histogram getOkLatencies() {
     return okLatencies;
   }
 
@@ -199,7 +199,7 @@ Histogram getOkLatencies() {
    * {@link SerializingMetricWriter}.
    */
   @Nullable
-  Histogram getErrorLatencies() {
+  public Histogram getErrorLatencies() {
     return errorLatencies;
   }
 
@@ -286,19 +286,19 @@ static long hashOf(
   }
 
   // Accessors for SerializingMetricWriter.
-  UTF8BytesString getResource() {
+  public UTF8BytesString getResource() {
     return resource;
   }
 
-  UTF8BytesString getService() {
+  public UTF8BytesString getService() {
     return service;
   }
 
-  UTF8BytesString getOperationName() {
+  public UTF8BytesString getOperationName() {
     return operationName;
   }
 
-  UTF8BytesString getServiceSource() {
+  public UTF8BytesString getServiceSource() {
     return serviceSource;
   }
 
@@ -308,41 +308,41 @@ UTF8BytesString getServiceSource() {
    * captured" (see field comment) -- callers that need a presence check should go through this
    * predicate rather than comparing against {@code EMPTY} directly.
    */
-  boolean hasServiceSource() {
+  public boolean hasServiceSource() {
     return serviceSource.length() > 0;
   }
 
-  UTF8BytesString getType() {
+  public UTF8BytesString getType() {
     return type;
   }
 
-  UTF8BytesString getSpanKind() {
+  public UTF8BytesString getSpanKind() {
     return spanKind;
   }
 
-  UTF8BytesString getHttpMethod() {
+  public UTF8BytesString getHttpMethod() {
     return httpMethod;
   }
 
   /**
    * Whether the snapshot carried an HTTP method. See {@link #hasServiceSource} for the contract.
    */
-  boolean hasHttpMethod() {
+  public boolean hasHttpMethod() {
     return httpMethod.length() > 0;
   }
 
-  UTF8BytesString getHttpEndpoint() {
+  public UTF8BytesString getHttpEndpoint() {
     return httpEndpoint;
   }
 
   /**
    * Whether the snapshot carried an HTTP endpoint. See {@link #hasServiceSource} for the contract.
    */
-  boolean hasHttpEndpoint() {
+  public boolean hasHttpEndpoint() {
     return httpEndpoint.length() > 0;
   }
 
-  UTF8BytesString getGrpcStatusCode() {
+  public UTF8BytesString getGrpcStatusCode() {
     return grpcStatusCode;
   }
 
@@ -350,23 +350,23 @@ UTF8BytesString getGrpcStatusCode() {
    * Whether the snapshot carried a gRPC status code. See {@link #hasServiceSource} for the
    * contract.
    */
-  boolean hasGrpcStatusCode() {
+  public boolean hasGrpcStatusCode() {
     return grpcStatusCode.length() > 0;
   }
 
-  int getHttpStatusCode() {
+  public int getHttpStatusCode() {
     return httpStatusCode;
   }
 
-  boolean isSynthetics() {
+  public boolean isSynthetics() {
     return synthetic;
   }
 
-  boolean isTraceRoot() {
+  public boolean isTraceRoot() {
     return traceRoot;
   }
 
-  List getPeerTags() {
+  public List getPeerTags() {
     return peerTags;
   }
 
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsSenderFactory.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsSenderFactory.java
new file mode 100644
index 00000000000..af7c116cfce
--- /dev/null
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsSenderFactory.java
@@ -0,0 +1,43 @@
+package datadog.trace.core.otlp.metrics;
+
+import datadog.trace.api.Config;
+import datadog.trace.core.otlp.common.OtlpGrpcSender;
+import datadog.trace.core.otlp.common.OtlpHttpSender;
+import datadog.trace.core.otlp.common.OtlpSender;
+import javax.annotation.Nullable;
+
+/**
+ * Selects the {@link OtlpSender} for the configured OTLP metrics protocol. Shared by every OTLP
+ * metrics export path ({@code OtlpMetricsService} for OpenTelemetry-API metrics, {@code
+ * OtlpStatsMetricWriter} for trace stats) so they all pick their transport identically and stay in
+ * sync as protocols/endpoints evolve.
+ */
+final class OtlpMetricsSenderFactory {
+  private OtlpMetricsSenderFactory() {}
+
+  /**
+   * Builds the sender for {@code config}'s OTLP metrics protocol, or {@code null} if the protocol
+   * has no protobuf encoder yet (e.g. HTTP_JSON).
+   */
+  @Nullable
+  static OtlpSender create(Config config) {
+    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:
+        return null;
+    }
+  }
+}
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
index ea6e28f47f9..5a1c0125dd0 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
@@ -3,8 +3,6 @@
 import static datadog.trace.util.AgentThreadFactory.AgentThread.OTLP_METRICS_EXPORTER;
 
 import datadog.trace.api.Config;
-import datadog.trace.core.otlp.common.OtlpGrpcSender;
-import datadog.trace.core.otlp.common.OtlpHttpSender;
 import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
 import datadog.trace.util.AgentTaskScheduler;
@@ -30,31 +28,12 @@ public final class OtlpMetricsService {
   private OtlpMetricsService(Config config) {
     this.scheduler = new AgentTaskScheduler(OTLP_METRICS_EXPORTER);
 
-    switch (config.getOtlpMetricsProtocol()) {
-      case GRPC:
-        this.collector = OtlpMetricsProtoCollector.INSTANCE;
-        this.sender =
-            new OtlpGrpcSender(
-                config.getOtlpMetricsEndpoint(),
-                "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export",
-                config.getOtlpMetricsHeaders(),
-                config.getOtlpMetricsTimeout(),
-                config.getOtlpMetricsCompression());
-        break;
-      case HTTP_PROTOBUF:
-        this.collector = OtlpMetricsProtoCollector.INSTANCE;
-        this.sender =
-            new OtlpHttpSender(
-                config.getOtlpMetricsEndpoint(),
-                "/v1/metrics",
-                config.getOtlpMetricsHeaders(),
-                config.getOtlpMetricsTimeout(),
-                config.getOtlpMetricsCompression());
-        break;
-      default:
-        LOGGER.debug("Unsupported OTLP metrics protocol: {}", config.getOtlpMetricsProtocol());
-        this.collector = null;
-        this.sender = null;
+    this.sender = OtlpMetricsSenderFactory.create(config);
+    if (this.sender == null) {
+      LOGGER.debug("Unsupported OTLP metrics protocol: {}", config.getOtlpMetricsProtocol());
+      this.collector = null;
+    } else {
+      this.collector = OtlpMetricsProtoCollector.INSTANCE;
     }
 
     this.intervalMillis = config.getMetricsOtelInterval();
diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsHistogramBuckets.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBuckets.java
similarity index 98%
rename from dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsHistogramBuckets.java
rename to dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBuckets.java
index 372cac73bae..87768705a1a 100644
--- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsHistogramBuckets.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBuckets.java
@@ -1,4 +1,4 @@
-package datadog.trace.common.metrics;
+package datadog.trace.core.otlp.metrics;
 
 import datadog.metrics.api.Histogram;
 import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint;
diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
similarity index 84%
rename from dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java
rename to dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
index 98256d61d4f..dc1814ce45c 100644
--- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/OtlpStatsMetricWriter.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
@@ -1,4 +1,4 @@
-package datadog.trace.common.metrics;
+package datadog.trace.core.otlp.metrics;
 
 import static datadog.trace.bootstrap.otel.metrics.OtelInstrumentType.HISTOGRAM;
 import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.LONG_ATTRIBUTE;
@@ -20,8 +20,8 @@
 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.common.metrics.AggregateEntry;
+import datadog.trace.common.metrics.MetricWriter;
 import datadog.trace.core.otlp.common.OtlpProtoBuffer;
 import datadog.trace.core.otlp.common.OtlpSender;
 import javax.annotation.Nullable;
@@ -33,8 +33,8 @@
  * 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
+ * 
This is the parallel-to-{@code SerializingMetricWriter} OTLP export path. It hangs off the
+ * same in-memory aggregation ({@code ClientStatsAggregator} / {@code 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).
  *
@@ -84,7 +84,15 @@ public final class OtlpStatsMetricWriter implements MetricWriter {
   private int metricBytes;
 
   public OtlpStatsMetricWriter(Config config) {
-    this(createSender(config), config.isTraceOtelSemanticsEnabled());
+    // shared protocol-based sender selection so both OTLP metrics export paths agree
+    this(OtlpMetricsSenderFactory.create(config), config.isTraceOtelSemanticsEnabled());
+    if (this.sender == null) {
+      // 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());
+    }
   }
 
   // visible for testing: lets tests inject a capturing sender to decode the emitted protobuf and
@@ -94,34 +102,6 @@ public OtlpStatsMetricWriter(Config config) {
     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)
diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java
index a8ced924c3e..60e0badcb7f 100644
--- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java
+++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/AggregateEntryTestUtils.java
@@ -89,6 +89,30 @@ public static AggregateEntry of(
         peerTagsList);
   }
 
+  /**
+   * Records one OK hit of {@code durationNanos} on {@code e}. Exposes the package-private {@link
+   * AggregateEntry#recordOneDuration} to tests in other packages (e.g. the OTLP metrics writer
+   * tests) without widening the production mutation API.
+   */
+  public static AggregateEntry recordOk(AggregateEntry e, long durationNanos) {
+    return e.recordOneDuration(durationNanos);
+  }
+
+  /** Records one error hit of {@code durationNanos} on {@code e}. See {@link #recordOk}. */
+  public static AggregateEntry recordError(AggregateEntry e, long durationNanos) {
+    return e.recordOneDuration(durationNanos | AggregateEntry.ERROR_TAG);
+  }
+
+  /** Records one top-level OK hit of {@code durationNanos} on {@code e}. See {@link #recordOk}. */
+  public static AggregateEntry recordTopLevel(AggregateEntry e, long durationNanos) {
+    return e.recordOneDuration(durationNanos | AggregateEntry.TOP_LEVEL_TAG);
+  }
+
+  /** Clears the per-cycle counters and histograms on {@code e}. See {@link #recordOk}. */
+  public static void clear(AggregateEntry e) {
+    e.clear();
+  }
+
   /**
    * Whether {@code a} and {@code b} carry identical label fields. Counter and histogram state is
    * intentionally excluded -- this compares the key identity, not the aggregate.
diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBucketsTest.java
similarity index 93%
rename from dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java
rename to dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBucketsTest.java
index d37cde2d92a..a24fe5d64b1 100644
--- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsHistogramBucketsTest.java
+++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBucketsTest.java
@@ -1,9 +1,9 @@
-package datadog.trace.common.metrics;
+package datadog.trace.core.otlp.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 datadog.trace.core.otlp.metrics.OtlpStatsHistogramBuckets.BOUNDS_SECONDS;
+import static datadog.trace.core.otlp.metrics.OtlpStatsHistogramBuckets.EXPLICIT_BOUNDS;
+import static datadog.trace.core.otlp.metrics.OtlpStatsHistogramBuckets.bucketIndex;
+import static datadog.trace.core.otlp.metrics.OtlpStatsHistogramBuckets.toHistogramPoint;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
similarity index 95%
rename from dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java
rename to dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
index 31ed707cabd..e7c90f20b33 100644
--- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/OtlpStatsMetricWriterTest.java
+++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
@@ -1,4 +1,4 @@
-package datadog.trace.common.metrics;
+package datadog.trace.core.otlp.metrics;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -12,6 +12,8 @@
 import com.google.protobuf.WireFormat;
 import datadog.metrics.api.Histograms;
 import datadog.metrics.impl.DDSketchHistograms;
+import datadog.trace.common.metrics.AggregateEntry;
+import datadog.trace.common.metrics.AggregateEntryTestUtils;
 import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
 import java.io.IOException;
@@ -103,7 +105,7 @@ private static AggregateEntry entry(
   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);
+      AggregateEntryTestUtils.recordOk(e, durationNanos);
     }
     return e;
   }
@@ -309,9 +311,9 @@ void okOnlyEntryProducesExactlyOneDataPoint() throws IOException {
   @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
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1)); // ok
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(2)); // ok
+    AggregateEntryTestUtils.recordError(e, SECONDS.toNanos(3)); // error
 
     DecodedMetric metric = writeAndDecode(false, e);
     assertEquals(2, metric.dataPoints.size(), "ok+error → two data points");
@@ -342,8 +344,8 @@ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOExcept
 
     // 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
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1)); // ok
+    AggregateEntryTestUtils.recordError(e, SECONDS.toNanos(3)); // error
 
     writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
     writer.add(e);
@@ -357,8 +359,8 @@ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOExcept
 
     // 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
+    AggregateEntryTestUtils.clear(e);
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(2)); // ok only
 
     writer.startBucket(2, BUCKET_START + BUCKET_DURATION, BUCKET_DURATION);
     writer.add(e);
@@ -373,7 +375,7 @@ void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOExcept
   @Test
   void httpAndGrpcAttributesAppearOnlyWhenSet() throws IOException {
     AggregateEntry e = entry("GET /users/{id}", false, 200, "GET", "/users/{id}", "0");
-    e.recordOneDuration(SECONDS.toNanos(1));
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
 
     DecodedMetric metric = writeAndDecode(false, e);
     assertEquals(1, metric.dataPoints.size());
@@ -422,7 +424,7 @@ void nullSenderDoesNotThrowOnNonEmptyBucket() {
   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);
+    AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
 
     Map attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
     assertTrue(
@@ -448,7 +450,7 @@ void defaultModeCarriesDatadogAttributes() throws IOException {
   void defaultModeEmitsSyntheticOrigin(boolean synthetic, String expectedOrigin)
       throws IOException {
     AggregateEntry e = entry("servlet.request", synthetic, 0, null, null, null);
-    e.recordOneDuration(SECONDS.toNanos(1));
+    AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1));
 
     Map attrs = writeAndDecode(false, e).dataPoints.get(0).attributes;
     if (expectedOrigin == null) {

From 8038e6ea159a5a60e72090fe93c8a94b0ab7ac56 Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Thu, 9 Jul 2026 16:33:52 -0400
Subject: [PATCH 15/16] wire otlp trace metrics thru OtlpMetricsProtoCollector

---
 .../metrics/OtlpMetricsProtoCollector.java    |  36 +++-
 .../core/otlp/metrics/OtlpMetricsService.java |   3 +-
 .../otlp/metrics/OtlpStatsMetricWriter.java   | 204 +++++++++---------
 .../metrics/OtlpStatsMetricWriterTest.java    |  28 +++
 4 files changed, 160 insertions(+), 111 deletions(-)

diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProtoCollector.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProtoCollector.java
index bf819668733..32e505fec8e 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProtoCollector.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProtoCollector.java
@@ -14,7 +14,6 @@
 import static datadog.trace.core.otlp.metrics.OtlpMetricsProto.recordScopedMetricsMessage;
 
 import datadog.communication.serialization.GrowableBuffer;
-import datadog.trace.api.time.SystemTimeSource;
 import datadog.trace.api.time.TimeSource;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.bootstrap.otel.metrics.OtelInstrumentDescriptor;
@@ -46,14 +45,13 @@
 public final class OtlpMetricsProtoCollector extends OtlpMetricsCollector
     implements OtlpMetricsVisitor, OtlpScopedMetricsVisitor, OtlpMetricVisitor {
 
-  public static final OtlpMetricsProtoCollector INSTANCE =
-      new OtlpMetricsProtoCollector(SystemTimeSource.INSTANCE);
-
   private final GrowableBuffer buf = new GrowableBuffer(512);
   private final OtlpProtoBuffer protobuf = new OtlpProtoBuffer(8192);
 
   private final TimeSource timeSource;
 
+  private final boolean forceHistogramDelta;
+
   private long startNanos;
   private long endNanos;
 
@@ -66,8 +64,13 @@ public final class OtlpMetricsProtoCollector extends OtlpMetricsCollector
   private OtelInstrumentDescriptor currentMetric;
 
   public OtlpMetricsProtoCollector(TimeSource timeSource) {
+    this(timeSource, false);
+  }
+
+  OtlpMetricsProtoCollector(TimeSource timeSource, boolean forceHistogramDelta) {
     this.timeSource = timeSource;
     this.endNanos = timeSource.getCurrentTimeNanos();
+    this.forceHistogramDelta = forceHistogramDelta;
   }
 
   /**
@@ -82,6 +85,16 @@ public OtlpPayload collectMetrics() {
 
   OtlpPayload collectMetrics(Consumer registry) {
     start();
+    return run(registry);
+  }
+
+  OtlpPayload collectMetrics(
+      Consumer registry, long startNanos, long endNanos) {
+    startWithWindow(startNanos, endNanos);
+    return run(registry);
+  }
+
+  private OtlpPayload run(Consumer registry) {
     try {
       registry.accept(this);
       return completePayload();
@@ -96,6 +109,18 @@ private void start() {
     startNanos = endNanos;
     endNanos = timeSource.getCurrentTimeNanos();
 
+    recalibrate();
+  }
+
+  /** Prepare temporary elements to collect metrics over an explicit window. */
+  private void startWithWindow(long startNanos, long endNanos) {
+    this.startNanos = startNanos;
+    this.endNanos = endNanos;
+
+    recalibrate();
+  }
+
+  private void recalibrate() {
     // remove stale entries from caches
     OtlpCommonProto.recalibrateCaches();
   }
@@ -193,7 +218,8 @@ private void completeMetric() {
 
     // add metric message prefix to its nested chunks and promote to scoped
     if (metricBytes > 0) {
-      scopedBytes += recordMetricMessage(buf, currentMetric, metricBytes, protobuf);
+      scopedBytes +=
+          recordMetricMessage(buf, currentMetric, metricBytes, protobuf, forceHistogramDelta);
     }
 
     // reset temporary elements for next metric
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
index 5a1c0125dd0..76d43104bdf 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsService.java
@@ -3,6 +3,7 @@
 import static datadog.trace.util.AgentThreadFactory.AgentThread.OTLP_METRICS_EXPORTER;
 
 import datadog.trace.api.Config;
+import datadog.trace.api.time.SystemTimeSource;
 import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
 import datadog.trace.util.AgentTaskScheduler;
@@ -33,7 +34,7 @@ private OtlpMetricsService(Config config) {
       LOGGER.debug("Unsupported OTLP metrics protocol: {}", config.getOtlpMetricsProtocol());
       this.collector = null;
     } else {
-      this.collector = OtlpMetricsProtoCollector.INSTANCE;
+      this.collector = new OtlpMetricsProtoCollector(SystemTimeSource.INSTANCE);
     }
 
     this.intervalMillis = config.getMetricsOtelInterval();
diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
index dc1814ce45c..952a09715ad 100644
--- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
+++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java
@@ -3,42 +3,30 @@
 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.api.time.SystemTimeSource;
 import datadog.trace.bootstrap.otel.common.OtelInstrumentationScope;
 import datadog.trace.bootstrap.otel.metrics.OtelInstrumentDescriptor;
-import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint;
+import datadog.trace.bootstrap.otlp.metrics.OtlpDataPoint;
+import datadog.trace.bootstrap.otlp.metrics.OtlpMetricVisitor;
+import datadog.trace.bootstrap.otlp.metrics.OtlpMetricsVisitor;
 import datadog.trace.common.metrics.AggregateEntry;
 import datadog.trace.common.metrics.MetricWriter;
-import datadog.trace.core.otlp.common.OtlpProtoBuffer;
+import datadog.trace.core.otlp.common.OtlpPayload;
 import datadog.trace.core.otlp.common.OtlpSender;
+import java.util.ArrayList;
+import java.util.List;
 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-{@code SerializingMetricWriter} OTLP export path. It hangs off the
- * same in-memory aggregation ({@code ClientStatsAggregator} / {@code 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}
+ * A {@link MetricWriter} that exports the client-side trace metrics as a delta-temporality OTLP
+ * histogram ({@code traces.span.sdk.metrics.duration}, unit {@code s}), the OTLP-native alternative
+ * to {@code SerializingMetricWriter}'s msgpack. It transforms each {@link AggregateEntry} into an
+ * OTLP histogram data point and pushes it through the shared {@link OtlpMetricsProtoCollector}.
  */
 public final class OtlpStatsMetricWriter implements MetricWriter {
   private static final Logger log = LoggerFactory.getLogger(OtlpStatsMetricWriter.class);
@@ -51,10 +39,6 @@ public final class OtlpStatsMetricWriter implements MetricWriter {
   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";
@@ -72,16 +56,30 @@ public final class OtlpStatsMetricWriter implements MetricWriter {
   @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);
+  // own single-thread collector; forced to DELTA since trace-stats buckets are per-interval deltas
+  private final OtlpMetricsProtoCollector collector =
+      new OtlpMetricsProtoCollector(SystemTimeSource.INSTANCE, true);
+
+  // data points snapshotted during add(), replayed through the visitor in finishBucket()
+  private final List pending = new ArrayList<>();
 
   private long startNanos;
   private long endNanos;
 
-  private int payloadBytes;
-  private int scopedBytes;
-  private int metricBytes;
+  // Represent a single datapoint from Aggregator.add
+  private static final class PendingPoint {
+    final AggregateEntry entry;
+    final OtlpDataPoint point;
+    final boolean error;
+    final boolean allTopLevel;
+
+    PendingPoint(AggregateEntry entry, OtlpDataPoint point, boolean error, boolean allTopLevel) {
+      this.entry = entry;
+      this.point = point;
+      this.error = error;
+      this.allTopLevel = allTopLevel;
+    }
+  }
 
   public OtlpStatsMetricWriter(Config config) {
     // shared protocol-based sender selection so both OTLP metrics export paths agree
@@ -107,113 +105,109 @@ 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;
+    pending.clear();
   }
 
   @Override
   public void add(AggregateEntry entry) {
+    // Value gets wiped when Aggregator clears the entry. Need to save it here
+    boolean allTopLevel = entry.getTopLevelCount() == entry.getHitCount();
+
     Histogram okLatencies = entry.getOkLatencies();
     if (!okLatencies.isEmpty()) {
-      addDataPoint(entry, okLatencies, false);
+      pending.add(
+          new PendingPoint(
+              entry,
+              OtlpStatsHistogramBuckets.toHistogramPoint(okLatencies, entry.getOkDuration()),
+              false,
+              allTopLevel));
     }
 
     Histogram errorLatencies = entry.getErrorLatencies();
     if (errorLatencies != null && !errorLatencies.isEmpty()) {
-      addDataPoint(entry, errorLatencies, true);
+      pending.add(
+          new PendingPoint(
+              entry,
+              OtlpStatsHistogramBuckets.toHistogramPoint(errorLatencies, entry.getErrorDuration()),
+              true,
+              allTopLevel));
+    }
+  }
+
+  @Override
+  public void finishBucket() {
+    try {
+      if (pending.isEmpty() || sender == null) {
+        return;
+      }
+      OtlpPayload payload = collector.collectMetrics(this::emit, startNanos, endNanos);
+      if (payload != OtlpPayload.EMPTY) {
+        sender.send(payload);
+      }
+    } finally {
+      pending.clear();
     }
   }
 
-  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);
+  @Override
+  public void reset() {
+    pending.clear();
+  }
+
+  /**
+   * Pushes the buffered entries through the metric visitor: one OTLP histogram data point per
+   * non-empty ok/error latency series. Called by {@link OtlpMetricsProtoCollector#collectMetrics}
+   * with the collector itself as the visitor.
+   */
+  private void emit(OtlpMetricsVisitor visitor) {
+    OtlpMetricVisitor metric = visitor.visitScopedMetrics(SCOPE).visitMetric(METRIC_DESCRIPTOR);
+    for (PendingPoint p : pending) {
+      // attributes must precede the data point (OtlpMetricVisitor contract)
+      writeDataPointAttributes(metric, p.entry, p.error, p.allTopLevel);
+      metric.visitDataPoint(p.point);
+    }
   }
 
-  private void writeDataPointAttributes(AggregateEntry entry, boolean error) {
+  private void writeDataPointAttributes(
+      OtlpMetricVisitor metric, AggregateEntry entry, boolean error, boolean allTopLevel) {
     if (error) {
-      writeStringAttribute(STATUS_CODE, STATUS_CODE_ERROR);
+      writeStringAttribute(metric, STATUS_CODE, STATUS_CODE_ERROR);
     }
     // OTel semconv attrs are emitted in both modes
-    writeStringAttribute(SPAN_NAME, entry.getResource());
-    writeStringAttribute(SPAN_KIND, entry.getSpanKind());
+    writeStringAttribute(metric, SPAN_NAME, entry.getResource());
+    writeStringAttribute(metric, SPAN_KIND, entry.getSpanKind());
     if (entry.hasHttpMethod()) {
-      writeStringAttribute(HTTP_REQUEST_METHOD, entry.getHttpMethod());
+      writeStringAttribute(metric, HTTP_REQUEST_METHOD, entry.getHttpMethod());
     }
     if (entry.getHttpStatusCode() != 0) {
-      writeLongAttribute(HTTP_RESPONSE_STATUS_CODE, entry.getHttpStatusCode());
+      writeLongAttribute(metric, HTTP_RESPONSE_STATUS_CODE, entry.getHttpStatusCode());
     }
     if (entry.hasHttpEndpoint()) {
-      writeStringAttribute(HTTP_ROUTE, entry.getHttpEndpoint());
+      writeStringAttribute(metric, HTTP_ROUTE, entry.getHttpEndpoint());
     }
     if (entry.hasGrpcStatusCode()) {
-      writeStringAttribute(RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode());
+      writeStringAttribute(metric, 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);
+      writeStringAttribute(metric, DATADOG_OPERATION_NAME, entry.getOperationName());
+      writeStringAttribute(metric, DATADOG_SPAN_TYPE, entry.getType());
+      writeLongAttribute(metric, DATADOG_SPAN_TOP_LEVEL, allTopLevel ? 1L : 0L);
       if (entry.isSynthetics()) {
-        writeStringAttribute(DATADOG_ORIGIN, SYNTHETICS_ORIGIN);
+        writeStringAttribute(metric, DATADOG_ORIGIN, SYNTHETICS_ORIGIN);
       }
     }
   }
 
-  private void writeStringAttribute(String key, @Nullable UTF8BytesString value) {
+  // accepts both String literals and UTF8BytesString (both CharSequence); skips null values
+  private static void writeStringAttribute(
+      OtlpMetricVisitor metric, String key, @Nullable CharSequence 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();
+      metric.visitAttribute(STRING_ATTRIBUTE, key, value.toString());
     }
   }
 
-  @Override
-  public void reset() {
-    buf.reset();
-    protobuf.reset();
-    payloadBytes = 0;
-    scopedBytes = 0;
-    metricBytes = 0;
+  private static void writeLongAttribute(OtlpMetricVisitor metric, String key, long value) {
+    metric.visitAttribute(LONG_ATTRIBUTE, key, value);
   }
 }
diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
index e7c90f20b33..e7152c4fdb8 100644
--- a/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
+++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java
@@ -476,4 +476,32 @@ void otelSemanticsModeOmitsDatadogAttributes() throws IOException {
     // OTel-semconv attrs must still be present
     assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode");
   }
+
+  @Test
+  void snapshotsEntryDataBeforeAggregatorClearsIt() throws IOException {
+    // The aggregator clears each entry's per-interval data immediately after add() returns
+    // (Aggregator#report), before finishBucket() runs. The writer must snapshot the latency data
+    // (and the top-level count) at add() time; if it deferred reading to finishBucket() it would
+    // encode the already-cleared (empty, zero-count) entry.
+    CapturingSender sender = new CapturingSender();
+    OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false);
+
+    AggregateEntry e = entry("servlet.request", false, 0, null, null, null);
+    AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
+    AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
+    AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1));
+
+    writer.startBucket(1, BUCKET_START, BUCKET_DURATION);
+    writer.add(e);
+    AggregateEntryTestUtils.clear(e); // mimic Aggregator#report clearing right after add()
+    writer.finishBucket();
+
+    assertEquals(1, sender.sendCount, "cleared-after-add entry must still emit its snapshot");
+    DecodedMetric metric = decode(sender.lastPayload);
+    assertEquals(1, metric.dataPoints.size());
+    DataPoint dp = metric.dataPoints.get(0);
+    assertEquals(3L, dp.count, "count must reflect the pre-clear snapshot, not the cleared entry");
+    assertEquals(
+        1L, dp.attributes.get("datadog.span.top_level"), "all pre-clear hits were top-level");
+  }
 }

From 92c84282ae6c68b095d12b70e088796fdca51a2d Mon Sep 17 00:00:00 2001
From: Matthew Li 
Date: Wed, 15 Jul 2026 14:26:32 -0400
Subject: [PATCH 16/16] jacoco

---
 dd-trace-core/build.gradle | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dd-trace-core/build.gradle b/dd-trace-core/build.gradle
index 4343576e181..327770b971a 100644
--- a/dd-trace-core/build.gradle
+++ b/dd-trace-core/build.gradle
@@ -65,6 +65,7 @@ excludedClassesCoverage += [
   'datadog.trace.core.otlp.common.OtlpGrpcSender',
   // covered by OTLP system-tests
   'datadog.trace.core.otlp.logs.OtlpLogsService',
+  'datadog.trace.core.otlp.metrics.OtlpMetricsSenderFactory',
   'datadog.trace.core.otlp.metrics.OtlpMetricsService'
 ]