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' ] 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/OtlpMetricsProto.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java index 751649940ba..a92c0826340 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpMetricsProto.java @@ -60,6 +60,20 @@ public static int recordMetricMessage( OtelInstrumentDescriptor descriptor, int nestedDataPointBytes, OtlpProtoBuffer protobuf) { + return recordMetricMessage(buf, descriptor, nestedDataPointBytes, protobuf, false); + } + + /** + * Records a metric message after its nested data point messages have been recorded. When {@code + * forceDelta} is {@code true}, a histogram is encoded as DELTA regardless of the configured + * temporality preference (for sources whose data points are inherently per-interval deltas). + */ + public static int recordMetricMessage( + GrowableBuffer buf, + OtelInstrumentDescriptor descriptor, + int nestedDataPointBytes, + OtlpProtoBuffer protobuf, + boolean forceDelta) { writeTag(buf, 1, LEN_WIRE_TYPE); writeString(buf, descriptor.getName().getUtf8Bytes()); @@ -107,7 +121,7 @@ public static int recordMetricMessage( writeTag(buf, 9, LEN_WIRE_TYPE); writeVarInt(buf, nestedDataPointBytes + 2); writeTag(buf, 2, VARINT_WIRE_TYPE); - writeVarInt(buf, HISTOGRAM_TEMPORALITY); + writeVarInt(buf, forceDelta ? TEMPORALITY_DELTA : HISTOGRAM_TEMPORALITY); break; default: throw new IllegalArgumentException("Unknown instrument type: " + descriptor.getType()); diff --git a/dd-trace-core/src/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/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..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,8 +3,7 @@ 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.api.time.SystemTimeSource; import datadog.trace.core.otlp.common.OtlpPayload; import datadog.trace.core.otlp.common.OtlpSender; import datadog.trace.util.AgentTaskScheduler; @@ -30,31 +29,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 = new OtlpMetricsProtoCollector(SystemTimeSource.INSTANCE); } this.intervalMillis = config.getMetricsOtelInterval(); diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBuckets.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBuckets.java new file mode 100644 index 00000000000..87768705a1a --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBuckets.java @@ -0,0 +1,70 @@ +package datadog.trace.core.otlp.metrics; + +import datadog.metrics.api.Histogram; +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 OtlpStatsHistogramBuckets { + private OtlpStatsHistogramBuckets() {} + + 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. + */ + static OtlpHistogramPoint toHistogramPoint(Histogram histogram, long sumNanos) { + long[] bucketCounts = new long[BOUNDS_SECONDS.length + 1]; + + List binBoundaries = histogram.getBinBoundaries(); + List binCounts = histogram.getBinCounts(); + 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; + } + + List counts = new ArrayList<>(bucketCounts.length); + for (long count : bucketCounts) { + counts.add((double) count); + } + + double sumSeconds = sumNanos / NANOS_PER_SECOND; + double minSeconds = histogram.isEmpty() ? 0d : histogram.getMinValue() / NANOS_PER_SECOND; + double maxSeconds = histogram.isEmpty() ? 0d : histogram.getMaxValue() / NANOS_PER_SECOND; + + return new OtlpHistogramPoint( + histogram.getCount(), EXPLICIT_BOUNDS, counts, sumSeconds, minSeconds, maxSeconds); + } +} diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java new file mode 100644 index 00000000000..952a09715ad --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriter.java @@ -0,0 +1,213 @@ +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; +import static datadog.trace.bootstrap.otlp.common.OtlpAttributeVisitor.STRING_ATTRIBUTE; + +import datadog.metrics.api.Histogram; +import datadog.trace.api.Config; +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.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.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 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); + + 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 String SPAN_NAME = "span.name"; + private static final String SPAN_KIND = "span.kind"; + private static final String HTTP_REQUEST_METHOD = "http.request.method"; + private static final String HTTP_RESPONSE_STATUS_CODE = "http.response.status_code"; + private static final String HTTP_ROUTE = "http.route"; + private static final String RPC_RESPONSE_STATUS_CODE = "rpc.response.status_code"; + private static final String STATUS_CODE = "status.code"; + private static final String STATUS_CODE_ERROR = "ERROR"; + private static final String DATADOG_OPERATION_NAME = "datadog.operation.name"; + private static final String DATADOG_SPAN_TYPE = "datadog.span.type"; + private static final String DATADOG_SPAN_TOP_LEVEL = "datadog.span.top_level"; + private static final String DATADOG_ORIGIN = "datadog.origin"; + private static final String SYNTHETICS_ORIGIN = "synthetics"; + + @Nullable private final OtlpSender sender; + private final boolean otelSemanticsMode; + + // 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; + + // 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 + 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 + // control the semantics mode + OtlpStatsMetricWriter(@Nullable OtlpSender sender, boolean otelSemanticsMode) { + this.sender = sender; + this.otelSemanticsMode = otelSemanticsMode; + } + + @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; + 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()) { + pending.add( + new PendingPoint( + entry, + OtlpStatsHistogramBuckets.toHistogramPoint(okLatencies, entry.getOkDuration()), + false, + allTopLevel)); + } + + Histogram errorLatencies = entry.getErrorLatencies(); + if (errorLatencies != null && !errorLatencies.isEmpty()) { + 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(); + } + } + + @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( + OtlpMetricVisitor metric, AggregateEntry entry, boolean error, boolean allTopLevel) { + if (error) { + writeStringAttribute(metric, STATUS_CODE, STATUS_CODE_ERROR); + } + // OTel semconv attrs are emitted in both modes + writeStringAttribute(metric, SPAN_NAME, entry.getResource()); + writeStringAttribute(metric, SPAN_KIND, entry.getSpanKind()); + if (entry.hasHttpMethod()) { + writeStringAttribute(metric, HTTP_REQUEST_METHOD, entry.getHttpMethod()); + } + if (entry.getHttpStatusCode() != 0) { + writeLongAttribute(metric, HTTP_RESPONSE_STATUS_CODE, entry.getHttpStatusCode()); + } + if (entry.hasHttpEndpoint()) { + writeStringAttribute(metric, HTTP_ROUTE, entry.getHttpEndpoint()); + } + if (entry.hasGrpcStatusCode()) { + writeStringAttribute(metric, RPC_RESPONSE_STATUS_CODE, entry.getGrpcStatusCode()); + } + // Default (Datadog) mode: emit datadog.* per-point attributes + if (!otelSemanticsMode) { + 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(metric, DATADOG_ORIGIN, SYNTHETICS_ORIGIN); + } + } + } + + // 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) { + metric.visitAttribute(STRING_ATTRIBUTE, key, value.toString()); + } + } + + 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/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/core/otlp/metrics/OtlpStatsHistogramBucketsTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBucketsTest.java new file mode 100644 index 00000000000..a24fe5d64b1 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsHistogramBucketsTest.java @@ -0,0 +1,145 @@ +package datadog.trace.core.otlp.metrics; + +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; + +import datadog.metrics.api.Histogram; +import datadog.metrics.api.Histograms; +import datadog.metrics.impl.DDSketchHistograms; +import datadog.trace.bootstrap.otlp.metrics.OtlpHistogramPoint; +import java.util.stream.IntStream; +import java.util.stream.Stream; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; + +/** + * Unit tests for {@link OtlpStatsHistogramBuckets}, the helper that re-bins a nanosecond-valued + * DDSketch onto the fixed OTLP explicit-bounds histogram layout (in seconds). + */ +class OtlpStatsHistogramBucketsTest { + + private static final double NANOS_PER_SECOND = 1_000_000_000d; + + // Tolerance for double assertions on exactly-computed values (e.g. sum from sumNanos, integer + // counts). DDSketch-derived values like min/max use looser tolerances inline, since the sketch + // only preserves values to within its relative accuracy. + private static final double EPS = 1e-9; + + @BeforeAll + static void registerHistogramFactory() { + Histograms.register(DDSketchHistograms.FACTORY); + } + + // ── bucketIndex ────────────────────────────────────────────────────────── + + static Stream boundsWithIndex() { + return IntStream.range(0, BOUNDS_SECONDS.length) + .mapToObj(i -> Arguments.of(i, BOUNDS_SECONDS[i])); + } + + @ParameterizedTest(name = "value exactly on BOUNDS_SECONDS[{0}]={1} returns index {0}") + @MethodSource("boundsWithIndex") + void valueExactlyOnBoundReturnsThatIndex(int index, double bound) { + // <= semantics: a value exactly on a bound falls in that bound's bucket. + assertEquals(index, bucketIndex(bound)); + } + + @Test + void valueJustAboveSmallBoundFallsInNextBucket() { + // 0.002 is the first bound; a value just above it but below the second bound (0.004) + // must roll into the next index. + assertEquals(0, bucketIndex(0.002)); + assertEquals(1, bucketIndex(0.0020001)); + assertEquals(1, bucketIndex(0.004)); + } + + @Test + void valueAboveLargestBoundIsOverflow() { + // > 15s overflows to the final (16th) index. + assertEquals(BOUNDS_SECONDS.length, bucketIndex(15.0000001)); + assertEquals(BOUNDS_SECONDS.length, bucketIndex(100.0)); + // exactly 15 (the largest non-overflow bound) is the last non-overflow index. + assertEquals(BOUNDS_SECONDS.length - 1, bucketIndex(15.0)); + } + + // ── EXPLICIT_BOUNDS layout ──────────────────────────────────────────────── + + @Test + void explicitBoundsHas17EntriesEndingInInfinity() { + assertEquals(17, EXPLICIT_BOUNDS.size()); + assertEquals(Double.POSITIVE_INFINITY, EXPLICIT_BOUNDS.get(EXPLICIT_BOUNDS.size() - 1)); + } + + // ── toHistogramPoint ────────────────────────────────────────────────────── + + @Test + void toHistogramPointSummary() { + Histogram h = Histogram.newHistogram(); + // 1ns and 1ms are below the smallest bound (0.002s) and must land in bucket 0; 100ms → bucket + // 6, 3s → bucket 13. + long[] samplesNanos = { + 1L, + (long) (0.001 * NANOS_PER_SECOND), + (long) (0.1 * NANOS_PER_SECOND), + (long) (3.0 * NANOS_PER_SECOND) + }; + for (long s : samplesNanos) { + h.accept(s); + } + + // Use a sumNanos that deliberately differs from the sketch's implied sum to prove the + // returned sum comes from the ARGUMENT, not the sketch. + long sumNanos = 42L * (long) NANOS_PER_SECOND; + OtlpHistogramPoint point = toHistogramPoint(h, sumNanos); + + // 17 bucket counts (the EXPLICIT_BOUNDS layout itself is covered by its own test). + assertEquals(17, point.bucketCounts.size()); + + // total count == number of samples. + assertEquals(samplesNanos.length, (long) point.count); + + // max is the 3s sample (CollapsingLowestDenseStore collapses the LOWEST bins, so the top is + // preserved accurately). The exact 1ns min is NOT recoverable: over this wide value range the + // lowest bins collapse, so getMinValue reports the collapsed bin (sub-2ms here), not 1ns. The + // tiny sample isn't lost though — that's proven by the count and bucket-0 assertions below. + // DDSketch is relative-accuracy, so min/max use loose tolerances. + assertTrue( + point.min > 0 && point.min <= BOUNDS_SECONDS[0], "min collapses into bucket 0 range"); + assertEquals(3.0, point.max, 1e-2); + + // sum equals the sumNanos ARGUMENT converted to seconds, not the sketch sum. + assertEquals(42.0, point.sum, EPS); + + long bucketTotal = 0; + for (int i = 0; i < point.bucketCounts.size(); i++) { + long c = (long) point.bucketCounts.get(i).doubleValue(); + bucketTotal += c; + if (i == 0) { + assertEquals(2L, c, "1ns + 1ms both land in bucket 0 (<= 0.002s)"); + } + } + assertEquals(samplesNanos.length, bucketTotal); + } + + @Test + void emptyHistogramHasZeroMinMaxAndCount() { + Histogram h = Histogram.newHistogram(); + OtlpHistogramPoint point = toHistogramPoint(h, 0L); + + assertEquals(0.0, point.count, EPS); + assertEquals(0.0, point.min, EPS); + assertEquals(0.0, point.max, EPS); + long bucketTotal = 0; + for (double c : point.bucketCounts) { + bucketTotal += (long) c; + } + assertEquals(0L, bucketTotal); + } +} diff --git a/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java new file mode 100644 index 00000000000..e7152c4fdb8 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/core/otlp/metrics/OtlpStatsMetricWriterTest.java @@ -0,0 +1,507 @@ +package datadog.trace.core.otlp.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.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; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import javax.annotation.Nullable; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +/** + * Tests for {@link OtlpStatsMetricWriter}. Drives the writer through {@code startBucket} → {@code + * add} → {@code finishBucket} with a capturing {@link OtlpSender}, then decodes the emitted + * protobuf {@code ExportMetricsServiceRequest} ({@code MetricsData}) using protobuf's {@link + * CodedInputStream}, reusing the decode idioms from {@code OtlpMetricsProtoTest}. + * + *

Wire layout (OTLP metrics proto): + * + * 

+ *   MetricsData        { ResourceMetrics resource_metrics = 1; }
+ *   ResourceMetrics    { Resource resource = 1; ScopeMetrics scope_metrics = 2; }
+ *   ScopeMetrics       { InstrumentationScope scope = 1; Metric metrics = 2; }
+ *   Metric             { string name = 1; string unit = 3; Histogram histogram = 9; }
+ *   Histogram          { HistogramDataPoint data_points = 1; AggregationTemporality = 2; }
+ *   HistogramDataPoint { fixed64 start = 2; fixed64 time = 3; fixed64 count = 4; double sum = 5;
+ *                        fixed64 bucket_counts = 6; double explicit_bounds = 7;
+ *                        KeyValue attributes = 9; double min = 11; double max = 12; }
+ *
+ */ +class OtlpStatsMetricWriterTest { + + private static final int TEMPORALITY_DELTA = 1; + private static final long BUCKET_START = SECONDS.toNanos(1_700_000_000L); + private static final long BUCKET_DURATION = SECONDS.toNanos(10); + + @BeforeAll + static void registerHistogramFactory() { + Histograms.register(DDSketchHistograms.FACTORY); + } + + // ── capturing sender ────────────────────────────────────────────────────── + + private static final class CapturingSender implements OtlpSender { + int sendCount; + byte[] lastPayload; + + @Override + public void send(OtlpPayload payload) { + sendCount++; + java.nio.ByteBuffer content = payload.getContent(); + byte[] bytes = new byte[content.remaining()]; + content.get(bytes); + lastPayload = bytes; + } + + @Override + public void shutdown() {} + } + + // ── entry builders ────────────────────────────────────────────────────── + + private static AggregateEntry entry( + String resource, + boolean synthetic, + int httpStatusCode, + @Nullable String httpMethod, + @Nullable String httpEndpoint, + @Nullable String grpcStatusCode) { + return AggregateEntryTestUtils.of( + resource, + "web", + "servlet.request", + null, + "web", + httpStatusCode, + synthetic, + true, + "server", + null, + httpMethod, + httpEndpoint, + grpcStatusCode); + } + + /** Build an entry and record {@code hits} ok durations of {@code durationNanos} each. */ + private static AggregateEntry okEntry(long durationNanos, int hits) { + AggregateEntry e = entry("GET /users", false, 0, null, null, null); + for (int i = 0; i < hits; i++) { + AggregateEntryTestUtils.recordOk(e, durationNanos); + } + return e; + } + + // ── decode helpers (adapted from OtlpMetricsProtoTest) ────────────────────── + + /** + * A decoded histogram data point. Only the fields this test asserts on are decoded: the window + * timestamps, the total count, and the attributes. Per-bucket contents (bucket_counts, + * explicit_bounds), sum, min, and max are intentionally not decoded here. + */ + private static final class DataPoint { + long start; + long end; + long count; + final Map attributes = new HashMap<>(); + } + + /** A decoded metric: name, unit, temporality, and its histogram data points. */ + private static final class DecodedMetric { + String name; + String unit; + int temporality = -1; + final List dataPoints = new ArrayList<>(); + } + + /** Decodes a full {@code MetricsData} payload into the single histogram metric it carries. */ + private static DecodedMetric decode(byte[] payload) throws IOException { + CodedInputStream metricsData = CodedInputStream.newInstance(payload); + int metricsTag = metricsData.readTag(); + assertEquals(1, WireFormat.getTagFieldNumber(metricsTag), "MetricsData.resource_metrics = 1"); + CodedInputStream resourceMetrics = metricsData.readBytes().newCodedInput(); + assertTrue(metricsData.isAtEnd(), "expected exactly one ResourceMetrics"); + + DecodedMetric metric = null; + while (!resourceMetrics.isAtEnd()) { + int tag = resourceMetrics.readTag(); + int field = WireFormat.getTagFieldNumber(tag); + if (field == 2) { // ScopeMetrics + metric = parseScopeMetrics(resourceMetrics.readBytes().newCodedInput()); + } else { + resourceMetrics.skipField(tag); // Resource (field 1) etc. + } + } + assertNotNull(metric, "no ScopeMetrics found"); + return metric; + } + + private static DecodedMetric parseScopeMetrics(CodedInputStream scopeMetrics) throws IOException { + DecodedMetric metric = null; + while (!scopeMetrics.isAtEnd()) { + int tag = scopeMetrics.readTag(); + if (WireFormat.getTagFieldNumber(tag) == 2) { // Metric + metric = parseMetric(scopeMetrics.readBytes().newCodedInput()); + } else { + scopeMetrics.skipField(tag); + } + } + return metric; + } + + private static DecodedMetric parseMetric(CodedInputStream m) throws IOException { + DecodedMetric metric = new DecodedMetric(); + while (!m.isAtEnd()) { + int tag = m.readTag(); + switch (WireFormat.getTagFieldNumber(tag)) { + case 1: + metric.name = m.readString(); + break; + case 3: + metric.unit = m.readString(); + break; + case 9: // Histogram + parseHistogram(m.readBytes().newCodedInput(), metric); + break; + default: + m.skipField(tag); + } + } + return metric; + } + + private static void parseHistogram(CodedInputStream h, DecodedMetric metric) throws IOException { + while (!h.isAtEnd()) { + int tag = h.readTag(); + switch (WireFormat.getTagFieldNumber(tag)) { + case 1: // HistogramDataPoint (repeated) + metric.dataPoints.add(parseDataPoint(h.readBytes().newCodedInput())); + break; + case 2: // aggregation_temporality + metric.temporality = h.readEnum(); + break; + default: + h.skipField(tag); + } + } + } + + private static DataPoint parseDataPoint(CodedInputStream dp) throws IOException { + DataPoint p = new DataPoint(); + while (!dp.isAtEnd()) { + int tag = dp.readTag(); + switch (WireFormat.getTagFieldNumber(tag)) { + case 2: + p.start = dp.readFixed64(); + break; + case 3: + p.end = dp.readFixed64(); + break; + case 4: + p.count = dp.readFixed64(); + break; + case 9: // attributes (KeyValue) + readKeyValue(dp.readBytes().newCodedInput(), p.attributes); + break; + default: // sum, bucket_counts, explicit_bounds, min, max — not asserted here + dp.skipField(tag); + } + } + return p; + } + + /** + * Reads a {@code KeyValue} into {@code out}: key (field 1) → value. Value is an {@code AnyValue} + * (field 2); we decode string (field 1) and int (field 3) variants used by this writer. + */ + private static void readKeyValue(CodedInputStream kv, Map out) + throws IOException { + String key = null; + Object value = null; + while (!kv.isAtEnd()) { + int tag = kv.readTag(); + switch (WireFormat.getTagFieldNumber(tag)) { + case 1: + key = kv.readString(); + break; + case 2: // AnyValue + value = readAnyValue(kv.readBytes().newCodedInput()); + break; + default: + kv.skipField(tag); + } + } + if (key != null) { + out.put(key, value); + } + } + + private static Object readAnyValue(CodedInputStream any) throws IOException { + Object value = null; + while (!any.isAtEnd()) { + int tag = any.readTag(); + switch (WireFormat.getTagFieldNumber(tag)) { + case 1: // string_value + value = any.readString(); + break; + case 3: // int_value + value = any.readInt64(); + break; + default: + any.skipField(tag); + } + } + return value; + } + + // ── writer driver ───────────────────────────────────────────────────────── + + /** + * Drives the writer through one full {@code startBucket → add → finishBucket} cycle for {@code + * entry} over the fixed {@link #BUCKET_START}/{@link #BUCKET_DURATION} window, asserts that + * exactly one payload was sent, and returns the decoded metric. + */ + private static DecodedMetric writeAndDecode(boolean otelSemanticsMode, AggregateEntry entry) + throws IOException { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, otelSemanticsMode); + writer.startBucket(1, BUCKET_START, BUCKET_DURATION); + writer.add(entry); + writer.finishBucket(); + assertEquals(1, sender.sendCount, "exactly one payload sent"); + return decode(sender.lastPayload); + } + + // ── test cases ────────────────────────────────────────────────────────── + + @Test + void okOnlyEntryProducesExactlyOneDataPoint() throws IOException { + DecodedMetric metric = writeAndDecode(false, okEntry(SECONDS.toNanos(1), 3)); + + assertEquals("traces.span.sdk.metrics.duration", metric.name); + assertEquals("s", metric.unit); + assertEquals(TEMPORALITY_DELTA, metric.temporality, "histogram must be delta temporality"); + assertEquals(1, metric.dataPoints.size(), "ok-only → one data point"); + + DataPoint dp = metric.dataPoints.get(0); + assertEquals(BUCKET_START, dp.start, "start_time_unix_nano == startBucket start"); + assertEquals(BUCKET_START + BUCKET_DURATION, dp.end, "time_unix_nano == start + duration"); + assertEquals(3L, dp.count); + assertFalse(dp.attributes.containsKey("status.code"), "ok point carries no status.code"); + } + + @Test + void okPlusErrorEntryProducesTwoDataPointsWithErrorStatus() throws IOException { + AggregateEntry e = entry("GET /users", false, 0, null, null, null); + 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"); + + long okCount = 0; + long errorCount = 0; + DataPoint errorPoint = null; + DataPoint okPoint = null; + for (DataPoint dp : metric.dataPoints) { + if ("ERROR".equals(dp.attributes.get("status.code"))) { + errorPoint = dp; + errorCount = dp.count; + } else { + okPoint = dp; + okCount = dp.count; + } + } + assertNotNull(errorPoint, "one data point must carry status.code=ERROR"); + assertNotNull(okPoint, "one data point must omit status.code"); + assertEquals(e.getOkLatencies().getCount(), (double) okCount, 1e-9); + assertEquals(e.getErrorLatencies().getCount(), (double) errorCount, 1e-9); + } + + @Test + void errorSeriesDoesNotLingerAfterClearWhenBucketHasOnlyOkHits() throws IOException { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false); + + // Bucket 1: the entry sees an error, so its error histogram is allocated and emits a point. + AggregateEntry e = entry("GET /users", false, 0, null, null, null); + AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1)); // ok + AggregateEntryTestUtils.recordError(e, SECONDS.toNanos(3)); // 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. + AggregateEntryTestUtils.clear(e); + AggregateEntryTestUtils.recordOk(e, 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"); + AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1)); + + DecodedMetric metric = writeAndDecode(false, e); + assertEquals(1, metric.dataPoints.size()); + Map attrs = metric.dataPoints.get(0).attributes; + + assertEquals("GET", attrs.get("http.request.method")); + assertEquals(200L, attrs.get("http.response.status_code")); + assertEquals("/users/{id}", attrs.get("http.route")); + assertEquals("0", attrs.get("rpc.response.status_code")); + + // a bare entry has none of these + Map bareAttrs = + writeAndDecode(false, okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes; + assertFalse(bareAttrs.containsKey("http.request.method")); + assertFalse(bareAttrs.containsKey("http.response.status_code")); + assertFalse(bareAttrs.containsKey("http.route")); + assertFalse(bareAttrs.containsKey("rpc.response.status_code")); + } + + @Test + void emptyBucketSendsNothing() { + CapturingSender sender = new CapturingSender(); + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(sender, false); + + writer.startBucket(0, BUCKET_START, BUCKET_DURATION); + writer.finishBucket(); // no add() + + assertEquals(0, sender.sendCount, "empty bucket must not invoke send"); + assertNull(sender.lastPayload); + } + + @Test + void nullSenderDoesNotThrowOnNonEmptyBucket() { + // mirrors the HTTP_JSON path where createSender(config) returns null. + OtlpStatsMetricWriter writer = new OtlpStatsMetricWriter(null, false); + writer.startBucket(1, BUCKET_START, BUCKET_DURATION); + writer.add(okEntry(SECONDS.toNanos(1), 2)); + try { + writer.finishBucket(); + } catch (Exception ex) { + fail("finishBucket must not throw with a null sender, but threw: " + ex); + } + } + + @Test + void defaultModeCarriesDatadogAttributes() throws IOException { + // use an entry where all hits are top-level: OR in TOP_LEVEL_TAG + AggregateEntry e = entry("servlet.request", false, 0, null, null, null); + AggregateEntryTestUtils.recordTopLevel(e, SECONDS.toNanos(1)); + + Map attrs = writeAndDecode(false, e).dataPoints.get(0).attributes; + assertTrue( + attrs.containsKey("datadog.operation.name"), "operation name present in default mode"); + assertTrue(attrs.containsKey("datadog.span.type"), "span type present in default mode"); + assertTrue( + attrs.containsKey("datadog.span.top_level"), "span top-level present in default mode"); + assertEquals(1L, attrs.get("datadog.span.top_level"), "all hits top-level → 1"); + // OTel-semconv attrs are present in both modes + assertTrue(attrs.containsKey("span.name"), "span.name present in both modes"); + // datadog.origin presence/absence is covered by defaultModeEmitsSyntheticOrigin + } + + /** + * In default mode a synthetic entry emits {@code datadog.origin = "synthetics"}; a non-synthetic + * entry omits the attribute. Origin has collapsed to a boolean {@code synthetic} flag upstream, + * so {@code "synthetics"} is the only origin value that can reach the writer. + */ + @ParameterizedTest(name = "synthetic={0} → datadog.origin={1}") + @CsvSource( + nullValues = "NULL", + value = {"false, NULL", "true, synthetics"}) + void defaultModeEmitsSyntheticOrigin(boolean synthetic, String expectedOrigin) + throws IOException { + AggregateEntry e = entry("servlet.request", synthetic, 0, null, null, null); + AggregateEntryTestUtils.recordOk(e, SECONDS.toNanos(1)); + + Map attrs = writeAndDecode(false, e).dataPoints.get(0).attributes; + if (expectedOrigin == null) { + assertFalse(attrs.containsKey("datadog.origin"), "non-synthetic → datadog.origin absent"); + } else { + assertEquals(expectedOrigin, attrs.get("datadog.origin"), "synthetic → datadog.origin"); + } + } + + @Test + void otelSemanticsModeOmitsDatadogAttributes() throws IOException { + // otelSemanticsMode = true → datadog.* must be absent + Map attrs = + writeAndDecode(true, okEntry(SECONDS.toNanos(1), 1)).dataPoints.get(0).attributes; + assertFalse( + attrs.containsKey("datadog.operation.name"), + "operation name absent in otel-semantics mode"); + assertFalse(attrs.containsKey("datadog.span.type"), "span type absent in otel-semantics mode"); + assertFalse( + attrs.containsKey("datadog.span.top_level"), + "span top-level absent in otel-semantics mode"); + assertFalse(attrs.containsKey("datadog.origin"), "origin absent in otel-semantics mode"); + // OTel-semconv attrs must still be present + assertTrue(attrs.containsKey("span.name"), "span.name present even in otel-semantics mode"); + } + + @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"); + } +}