From 1d5d022a443bd209e07e86deffb90c20240ff3c1 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Thu, 18 Jun 2026 17:57:24 -0400 Subject: [PATCH 1/7] init --- .../common/metrics/ClientStatsAggregator.java | 25 ++++ .../metrics/MetricsAggregatorFactory.java | 8 ++ .../trace/common/metrics/NoOpSink.java | 23 +++ .../trace/common/writer/WriterFactory.java | 2 +- .../core/otlp/common/OtlpResourceProto.java | 29 +++- .../otlp/trace/OtlpTraceProtoCollector.java | 4 +- .../MetricsAggregatorFactoryTest.groovy | 33 ----- .../common/writer/WriterFactoryTest.groovy | 45 ++++++ .../metrics/MetricsAggregatorFactoryTest.java | 133 ++++++++++++++++++ .../otlp/common/OtlpResourceProtoTest.java | 26 +++- 10 files changed, 281 insertions(+), 47 deletions(-) create mode 100644 dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java delete mode 100644 dd-trace-core/src/test/groovy/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.groovy create mode 100644 dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 3a3ded31ff8..35d8a6b4380 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -12,6 +12,7 @@ import static datadog.trace.util.AgentThreadFactory.AgentThread.METRICS_AGGREGATOR; import static datadog.trace.util.AgentThreadFactory.THREAD_JOIN_TIMOUT_MS; import static datadog.trace.util.AgentThreadFactory.newAgentThread; +import static java.util.concurrent.TimeUnit.MILLISECONDS; import static java.util.concurrent.TimeUnit.SECONDS; import datadog.common.queue.Queues; @@ -114,6 +115,30 @@ public ClientStatsAggregator( config.isTraceResourceRenamingEnabled()); } + /** + * OTLP span-metrics export variant. Reuses the same span selection + DDSketch aggregation as the + * native path, but emits via the injected {@link OtlpStatsMetricWriter} instead of msgpack to. No + * agent {@link Sink} is needed, so a {@link NoOpSink} satisfies the register()/backpressure + * contract, and the reporting interval comes from {@code trace.stats.interval} (milliseconds). + */ + public ClientStatsAggregator( + Config config, + SharedCommunicationObjects sharedCommunicationObjects, + HealthMetrics healthMetrics, + OtlpStatsMetricWriter metricWriter) { + this( + config.getMetricsIgnoredResources(), + sharedCommunicationObjects.featuresDiscovery(config), + healthMetrics, + NoOpSink.INSTANCE, + metricWriter, + config.getTracerMetricsMaxAggregates(), + config.getTracerMetricsMaxPending(), + config.getTraceStatsInterval(), + MILLISECONDS, + config.isTraceResourceRenamingEnabled()); + } + ClientStatsAggregator( WellKnownTags wellKnownTags, Set ignoredResources, diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java index b9530871763..81de46c9113 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java @@ -13,6 +13,14 @@ public static MetricsAggregator createMetricsAggregator( Config config, SharedCommunicationObjects sharedCommunicationObjects, HealthMetrics healthMetrics) { + // OTLP span-metrics export and native msgpack stats are mutually exclusive (XOR): both hang off + // the same ClientStatsAggregator span selection + DDSketch aggregation, differing only in + // the injected MetricWriter. + if (config.isTracesSpanMetricsEnabled()) { + log.debug("OTLP trace span metrics enabled"); + return new ClientStatsAggregator( + config, sharedCommunicationObjects, healthMetrics, new OtlpStatsMetricWriter(config)); + } if (config.isTracerMetricsEnabled()) { log.debug("tracer metrics enabled"); return new ClientStatsAggregator(config, sharedCommunicationObjects, healthMetrics); diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java new file mode 100644 index 00000000000..f624f5f403c --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java @@ -0,0 +1,23 @@ +package datadog.trace.common.metrics; + +import java.nio.ByteBuffer; + +/** + * A {@link Sink} that discards everything. Used by the OTLP trace-metrics export path, where {@link + * OtlpStatsMetricWriter} sends payloads directly via its own OTLP sender in {@code finishBucket()} + * rather than through the aggregator's {@code Sink}. {@link ClientStatsAggregator} still + * requires a {@code Sink} for {@code register()}/backpressure wiring, so this satisfies that + * contract without performing any I/O. + */ +public final class NoOpSink implements Sink { + + public static final NoOpSink INSTANCE = new NoOpSink(); + + private NoOpSink() {} + + @Override + public void accept(int messageCount, ByteBuffer buffer) {} + + @Override + public void register(EventListener listener) {} +} diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java index 3f38d45881f..fc8a1ef00cc 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java @@ -187,7 +187,7 @@ public static Writer createWriter( commObjects.agentUrl, featuresDiscovery, commObjects.monitoring, - config.isTracerMetricsEnabled()); + config.isTracerMetricsEnabled() || config.isTracesSpanMetricsEnabled()); if (sampler instanceof RemoteResponseListener) { ddAgentApi.addResponseListener((RemoteResponseListener) sampler); 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 ffd2810a996..ef15009c75e 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 @@ -36,8 +36,14 @@ private OtlpResourceProto() {} /** Prefix applied to {@code datadog.runtime_id} and process-tag resource attributes. */ private static final String DATADOG_PREFIX = "datadog."; - /** Vendor-neutral resource (no {@code datadog.*}). Used by the OTLP trace/metric export. */ - public static final byte[] RESOURCE_MESSAGE = buildResourceMessage(Config.get()); + /** + * Resource attribute added to OTLP trace metrics to ensure calculations are not re-computed in + * the Agent + */ + private static final String STATS_COMPUTED_KEY = "_dd.stats_computed"; + + /** Vendor-neutral resource (no {@code datadog.*}). Used by the OTLP metric export. */ + public static final byte[] RESOURCE_MESSAGE = buildResourceMessage(Config.get(), false, false); /** * Resource that additionally carries {@code datadog.runtime_id} and process tags (each prefixed @@ -45,13 +51,18 @@ private OtlpResourceProto() {} * mode. */ public static final byte[] RESOURCE_MESSAGE_WITH_DATADOG_ATTRS = - buildResourceMessage(Config.get(), true); + buildResourceMessage(Config.get(), true, false); - static byte[] buildResourceMessage(Config config) { - return buildResourceMessage(config, false); - } + /** + * Resource used by the OTLP trace export. Identical to {@link #RESOURCE_MESSAGE} but adds the + * {@code _dd.stats_computed} marker when the SDK is computing OTLP span metrics, so a downstream + * Agent does not recompute them from the exported spans. + */ + public static final byte[] TRACE_RESOURCE_MESSAGE = + buildResourceMessage(Config.get(), false, Config.get().isTracesSpanMetricsEnabled()); - static byte[] buildResourceMessage(Config config, boolean includeDatadogResourceAttributes) { + static byte[] buildResourceMessage( + Config config, boolean includeDatadogResourceAttributes, boolean includeStatsComputed) { GrowableBuffer buf = new GrowableBuffer(512); String serviceName = config.getServiceName(); @@ -89,6 +100,10 @@ static byte[] buildResourceMessage(Config config, boolean includeDatadogResource writeDatadogResourceAttributes(buf, config); } + if (includeStatsComputed) { + writeResourceAttribute(buf, STATS_COMPUTED_KEY, "true"); + } + OtlpProtoBuffer protobuf = new OtlpProtoBuffer(buf.capacity()); int numBytes = protobuf.recordMessage(buf, 1); byte[] resourceMessage = new byte[numBytes]; diff --git a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java index f37b7a5aea2..fa1efa87fba 100644 --- a/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java +++ b/dd-trace-core/src/main/java/datadog/trace/core/otlp/trace/OtlpTraceProtoCollector.java @@ -1,6 +1,6 @@ package datadog.trace.core.otlp.trace; -import static datadog.trace.core.otlp.common.OtlpResourceProto.RESOURCE_MESSAGE; +import static datadog.trace.core.otlp.common.OtlpResourceProto.TRACE_RESOURCE_MESSAGE; import static datadog.trace.core.otlp.trace.OtlpTraceProto.recordScopedSpansMessage; import static datadog.trace.core.otlp.trace.OtlpTraceProto.recordSpanLinkMessage; import static datadog.trace.core.otlp.trace.OtlpTraceProto.recordSpanMessage; @@ -138,7 +138,7 @@ private OtlpPayload completePayload() { } // prepend the canned resource chunk - payloadBytes += protobuf.recordMessage(RESOURCE_MESSAGE); + payloadBytes += protobuf.recordMessage(TRACE_RESOURCE_MESSAGE); // finally prepend the total length of all collected chunks protobuf.recordMessage(buf, 1, payloadBytes); diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.groovy deleted file mode 100644 index dc9eb86fde3..00000000000 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.groovy +++ /dev/null @@ -1,33 +0,0 @@ -package datadog.trace.common.metrics - -import datadog.communication.ddagent.SharedCommunicationObjects -import datadog.trace.api.Config -import datadog.trace.core.monitor.HealthMetrics -import datadog.trace.test.util.DDSpecification -import okhttp3.HttpUrl - -class MetricsAggregatorFactoryTest extends DDSpecification { - - def "when metrics disabled no-op aggregator created"() { - setup: - Config config = Mock(Config) - config.isTracerMetricsEnabled() >> false - def sco = Mock(SharedCommunicationObjects) - sco.agentUrl = HttpUrl.parse("http://localhost:8126") - expect: - def aggregator = MetricsAggregatorFactory.createMetricsAggregator(config, sco, HealthMetrics.NO_OP,) - assert aggregator instanceof NoOpMetricsAggregator - } - - def "when metrics enabled conflating aggregator created"() { - setup: - Config config = Spy(Config.get()) - config.isTracerMetricsEnabled() >> true - def sco = Mock(SharedCommunicationObjects) - sco.agentUrl = HttpUrl.parse("http://localhost:8126") - expect: - def aggregator = MetricsAggregatorFactory.createMetricsAggregator(config, sco, HealthMetrics.NO_OP, - ) - assert aggregator instanceof ClientStatsAggregator - } -} diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy index d20bd475cac..59cc60c3820 100644 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy +++ b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy @@ -207,6 +207,51 @@ class WriterFactoryTest extends DDSpecification { OtlpConfig.Protocol.GRPC | OtlpConfig.Compression.NONE | "http://otel-collector:4317" | OtlpGrpcSender | "http://otel-collector:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export" | false } + def "DDAgentApi marks client-computed stats when either stats pipeline is enabled (spanMetrics=#spanMetrics, nativeStats=#nativeStats)"() { + setup: + // metricsEnabled gates the Datadog-Client-Computed-Stats header on the native trace transport. + // It must be true whenever the SDK computes stats by EITHER pipeline: native msgpack stats + // (isTracerMetricsEnabled) or OTLP span metrics (isTracesSpanMetricsEnabled). The OTLP-stats + // case arises when OTEL_TRACES_SPAN_METRICS_ENABLED is set while traces still export natively; + // without the header the Agent would recompute stats from spans already exported via OTLP. + def config = Mock(Config) + config.apiKey >> "my-api-key" + config.agentUrl >> "http://my-agent.url" + config.getEnumValue(PRIORITIZATION_TYPE, _, _) >> Prioritization.FAST_LANE + config.tracerMetricsEnabled >> nativeStats + config.tracesSpanMetricsEnabled >> spanMetrics + + def mockCall = Mock(Call) + def mockHttpClient = Mock(OkHttpClient) + // advertise only v0.4 (no EVP proxy) so "DDAgentWriter" resolves to a real DDAgentWriter + mockCall.execute() >> { buildHttpResponse(false, false, HttpUrl.parse(config.agentUrl + "/info")) } + mockHttpClient.newCall(_ as Request) >> mockCall + + def sharedComm = new SharedCommunicationObjects() + sharedComm.agentHttpClient = mockHttpClient + sharedComm.agentUrl = HttpUrl.parse(config.agentUrl) + sharedComm.createRemaining(config) + + def sampler = Mock(Sampler) + + when: + def writer = WriterFactory.createWriter(config, sharedComm, sampler, null, HealthMetrics.NO_OP, "DDAgentWriter") + def api = ((RemoteWriter) writer).apis.find { it instanceof DDAgentApi } + def metricsEnabledField = DDAgentApi.getDeclaredField("metricsEnabled") + metricsEnabledField.setAccessible(true) + + then: + writer instanceof DDAgentWriter + metricsEnabledField.getBoolean(api) == expectedComputesStats + + where: + spanMetrics | nativeStats | expectedComputesStats + true | false | true // gap case: OTLP span metrics on, native stats off + false | false | false // neither pipeline computes stats + false | true | true // native stats on (regression guard) + true | true | true // both on + } + Response buildHttpResponse(boolean hasEvpProxy, boolean evpProxySupportsCompression, HttpUrl agentUrl) { def endpoints = [] if (hasEvpProxy && evpProxySupportsCompression) { diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java new file mode 100644 index 00000000000..7453a9984c7 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java @@ -0,0 +1,133 @@ +package datadog.trace.common.metrics; + +import static datadog.trace.api.config.GeneralConfig.TRACE_STATS_COMPUTATION_ENABLED; +import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_PROTOCOL; +import static datadog.trace.api.config.OtlpConfig.TRACES_SPAN_METRICS_ENABLED; +import static datadog.trace.api.config.OtlpConfig.TRACE_OTEL_EXPORTER; +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.trace.api.Config; +import datadog.trace.core.monitor.HealthMetrics; +import java.lang.reflect.Field; +import java.util.Properties; +import java.util.concurrent.TimeUnit; +import okhttp3.HttpUrl; +import org.junit.jupiter.api.Test; + +/** + * Tests the native-vs-OTLP XOR writer selection in {@link MetricsAggregatorFactory}. The selected + * {@link MetricWriter} is not exposed publicly, so it is read via reflection through {@code + * ClientStatsAggregator.aggregator -> Aggregator.writer}. + */ +class MetricsAggregatorFactoryTest { + + private static SharedCommunicationObjects sharedCommunicationObjects() { + SharedCommunicationObjects sco = mock(SharedCommunicationObjects.class); + sco.agentUrl = HttpUrl.parse("http://localhost:8126"); + when(sco.featuresDiscovery(any())).thenReturn(mock(DDAgentFeaturesDiscovery.class)); + return sco; + } + + private static Properties props(String... keyValues) { + Properties props = new Properties(); + for (int i = 0; i < keyValues.length; i += 2) { + props.setProperty(keyValues[i], keyValues[i + 1]); + } + return props; + } + + @Test + void whenAllMetricsDisabledNoOpAggregatorCreated() { + Config config = Config.get(props(TRACE_STATS_COMPUTATION_ENABLED, "false")); + + MetricsAggregator aggregator = + MetricsAggregatorFactory.createMetricsAggregator( + config, sharedCommunicationObjects(), HealthMetrics.NO_OP); + + assertInstanceOf(NoOpMetricsAggregator.class, aggregator); + } + + @Test + void whenNativeTracerMetricsEnabledSerializingWriterSelected() throws Exception { + // tracer metrics default to enabled; OTLP span metrics default off (no OTLP trace export). + Config config = Config.get(props()); + + MetricsAggregator aggregator = + MetricsAggregatorFactory.createMetricsAggregator( + config, sharedCommunicationObjects(), HealthMetrics.NO_OP); + + assertInstanceOf(ClientStatsAggregator.class, aggregator); + assertInstanceOf(SerializingMetricWriter.class, writerOf(aggregator)); + } + + @Test + void whenOtlpSpanMetricsEnabledOtlpWriterSelectedWithConfiguredInterval() throws Exception { + // OTEL_TRACES_EXPORTER=otlp satisfies FR16 (traces not routed through an Agent). http/json has + // no protobuf-free encoder, so the writer's sender is null -- keeps construction lightweight + // (no + // real OTLP sender / network) while still exercising writer selection. + Config config = + Config.get( + props( + TRACES_SPAN_METRICS_ENABLED, "true", + TRACE_OTEL_EXPORTER, "otlp", + OTLP_METRICS_PROTOCOL, "http/json")); + + MetricsAggregator aggregator = + MetricsAggregatorFactory.createMetricsAggregator( + config, sharedCommunicationObjects(), HealthMetrics.NO_OP); + + assertInstanceOf(ClientStatsAggregator.class, aggregator); + assertInstanceOf(OtlpStatsMetricWriter.class, writerOf(aggregator)); + // reporting interval comes from getTraceStatsInterval() (ms), default 10s. + assertEquals(config.getTraceStatsInterval(), reportingIntervalOf(aggregator)); + assertEquals(MILLISECONDS, reportingIntervalUnitOf(aggregator)); + } + + @Test + void explicitSpanMetricsEnabledSelectsOtlpWriterEvenWithoutOtlpTraceExport() throws Exception { + // An explicit OTEL_TRACES_SPAN_METRICS_ENABLED=true is honored verbatim regardless of trace + // exporter (matching the dd-trace-py / dd-trace-go reference impls). Double-counting if these + // OTLP spans reach an Agent is handled by the FR15 _dd.stats_computed marker, not by disabling + // here. http/json keeps the writer's sender null (no network at construction). + Config config = + Config.get(props(TRACES_SPAN_METRICS_ENABLED, "true", OTLP_METRICS_PROTOCOL, "http/json")); + + MetricsAggregator aggregator = + MetricsAggregatorFactory.createMetricsAggregator( + config, sharedCommunicationObjects(), HealthMetrics.NO_OP); + + assertInstanceOf(ClientStatsAggregator.class, aggregator); + assertInstanceOf(OtlpStatsMetricWriter.class, writerOf(aggregator)); + } + + // ── reflection helpers ───────────────────────────────────────────────────── + + private static MetricWriter writerOf(MetricsAggregator aggregator) throws Exception { + Field aggregatorField = ClientStatsAggregator.class.getDeclaredField("aggregator"); + aggregatorField.setAccessible(true); + Object inner = aggregatorField.get(aggregator); + Field writerField = Aggregator.class.getDeclaredField("writer"); + writerField.setAccessible(true); + return (MetricWriter) writerField.get(inner); + } + + private static long reportingIntervalOf(MetricsAggregator aggregator) throws Exception { + Field f = ClientStatsAggregator.class.getDeclaredField("reportingInterval"); + f.setAccessible(true); + return f.getLong(aggregator); + } + + private static TimeUnit reportingIntervalUnitOf(MetricsAggregator aggregator) throws Exception { + Field f = ClientStatsAggregator.class.getDeclaredField("reportingIntervalTimeUnit"); + f.setAccessible(true); + return (TimeUnit) f.get(aggregator); + } +} 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 80740dc7e54..5978398d0ec 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 @@ -149,14 +149,14 @@ void testBuildResourceMessage( String caseName, Properties properties, Map expectedAttributes) throws IOException { Config config = Config.get(properties); - byte[] bytes = OtlpResourceProto.buildResourceMessage(config); + byte[] bytes = OtlpResourceProto.buildResourceMessage(config, false, false); Map actualAttributes = parseResourceAttributes(bytes); assertEquals(expectedAttributes, actualAttributes, "For case: " + caseName); } /** - * The datadog-attrs variant ({@code buildResourceMessage(config, true)}) carries {@code + * The datadog-attrs variant ({@code buildResourceMessage(config, true, false)}) carries {@code * datadog.runtime_id}; the plain variant omits it. (Process tags are emitted only when the * experimental process-tag propagation is enabled, so they aren't asserted here.) */ @@ -165,9 +165,9 @@ void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException { Config config = Config.get(props(SERVICE_NAME, "my-service")); Map withDatadog = - parseResourceAttributes(OtlpResourceProto.buildResourceMessage(config, true)); + parseResourceAttributes(OtlpResourceProto.buildResourceMessage(config, true, false)); Map plain = - parseResourceAttributes(OtlpResourceProto.buildResourceMessage(config, false)); + parseResourceAttributes(OtlpResourceProto.buildResourceMessage(config, false, false)); assertTrue( withDatadog.containsKey("datadog.runtime_id"), @@ -179,6 +179,24 @@ void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException { assertFalse(plain.containsKey("datadog.runtime_id"), "plain variant omits datadog.runtime_id"); } + /** + * FR15: the {@code includeStatsComputed} variant adds the {@code _dd.stats_computed=true} marker + * (used on the OTLP trace payload so a downstream Agent skips recompute); the default omits it. + */ + @Test + void statsComputedVariantCarriesMarker() throws IOException { + Config config = Config.get(props(SERVICE_NAME, "my-service")); + + Map withMarker = + parseResourceAttributes(OtlpResourceProto.buildResourceMessage(config, false, true)); + Map without = + parseResourceAttributes(OtlpResourceProto.buildResourceMessage(config, false, false)); + + assertEquals( + "true", withMarker.get("_dd.stats_computed"), "marker present when stats computed"); + assertFalse(without.containsKey("_dd.stats_computed"), "marker absent when stats not computed"); + } + // ── parsing helpers ─────────────────────────────────────────────────────── /** From b1b311ba4e7ace6cb07ac89a174f0c4343437bfe Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Tue, 23 Jun 2026 09:36:44 -0400 Subject: [PATCH 2/7] adding bugfixes for system-tests --- .../ddagent/DDAgentFeaturesDiscovery.java | 3 ++- .../common/metrics/ClientStatsAggregator.java | 20 +++++++++++++++---- .../metrics/MetricsAggregatorFactory.java | 6 ++++++ .../trace/common/writer/DDAgentWriter.java | 12 ++++++++++- .../trace/common/writer/WriterFactory.java | 8 ++++++++ .../common/writer/ddagent/DDAgentApi.java | 3 ++- 6 files changed, 45 insertions(+), 7 deletions(-) diff --git a/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java b/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java index 7cef3e02924..16be2e84b98 100644 --- a/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java +++ b/communication/src/main/java/datadog/communication/ddagent/DDAgentFeaturesDiscovery.java @@ -185,10 +185,11 @@ private void doDiscovery(State newState) { if (log.isDebugEnabled()) { log.debug( - "discovered traceEndpoint={}, metricsEndpoint={}, supportsDropping={}, supportsLongRunning={}, dataStreamsEndpoint={}, configEndpoint={}, logEndpoint={}, snapshotEndpoint={}, diagnosticsEndpoint={}, evpProxyEndpoint={}, telemetryProxyEndpoint={}", + "discovered traceEndpoint={}, metricsEndpoint={}, supportsDropping={}, supportsClientSideStats={}, supportsLongRunning={}, dataStreamsEndpoint={}, configEndpoint={}, logEndpoint={}, snapshotEndpoint={}, diagnosticsEndpoint={}, evpProxyEndpoint={}, telemetryProxyEndpoint={}", newState.traceEndpoint, newState.metricsEndpoint, newState.supportsDropping, + newState.supportsClientSideStats, newState.supportsLongRunning, newState.dataStreamsEndpoint, newState.configEndpoint, diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 35d8a6b4380..f8da68407dc 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -4,6 +4,7 @@ import static datadog.trace.api.DDSpanTypes.RPC; import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ENDPOINT; import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_METHOD; +import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ROUTE; import static datadog.trace.common.metrics.AggregateEntry.ERROR_TAG; import static datadog.trace.common.metrics.AggregateEntry.TOP_LEVEL_TAG; import static datadog.trace.common.metrics.SignalItem.ClearSignal.CLEAR; @@ -73,6 +74,7 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList private final DDAgentFeaturesDiscovery features; private final HealthMetrics healthMetrics; private final boolean includeEndpointInMetrics; + private final boolean otlpStatsExportEnabled; /** * Cached peer-aggregation schema read by producer threads. @@ -136,7 +138,7 @@ public ClientStatsAggregator( config.getTracerMetricsMaxPending(), config.getTraceStatsInterval(), MILLISECONDS, - config.isTraceResourceRenamingEnabled()); + true); } ClientStatsAggregator( @@ -198,6 +200,7 @@ public ClientStatsAggregator( boolean includeEndpointInMetrics) { this.ignoredResources = ignoredResources; this.includeEndpointInMetrics = includeEndpointInMetrics; + this.otlpStatsExportEnabled = metricWriter instanceof OtlpStatsMetricWriter; this.inbox = Queues.mpscArrayQueue(queueSize); this.features = features; this.healthMetrics = healthMetric; @@ -231,11 +234,19 @@ public void start() { log.debug("started metrics aggregator"); } + private boolean statsExportEnabled() { + return otlpStatsExportEnabled || features.supportsMetrics(); + } + private boolean isMetricsEnabled() { - if (features.getMetricsEndpoint() == null) { + // The discovery refresh only helps the native path, which is gated on the agent advertising + // v0.6/stats. The OTLP path uses its own sender and never depends on that capability -- even + // when a Datadog Agent is present (the OTLP endpoint may itself be the agent's OTLP receiver), + // so refreshing agent features here would be pointless for it. + if (!otlpStatsExportEnabled && features.getMetricsEndpoint() == null) { features.discoverIfOutdated(); } - return features.supportsMetrics(); + return statsExportEnabled(); } @Override @@ -291,7 +302,7 @@ public Future forceReport() { public boolean publish(List> trace) { boolean forceKeep = false; int counted = 0; - if (features.supportsMetrics()) { + if (statsExportEnabled()) { // Producer-side fast path: one volatile read and use whatever schema is currently cached. // The aggregator thread keeps this schema in sync with feature discovery in // resetCardinalityHandlers(). The only producer-side rebuild is the one-time bootstrap on @@ -340,6 +351,7 @@ private boolean publish(CoreSpan span, boolean isTopLevel, PeerTagSchema peer Object httpMethodObj = span.unsafeGetTag(HTTP_METHOD); httpMethod = httpMethodObj != null ? httpMethodObj.toString() : null; Object httpEndpointObj = span.unsafeGetTag(HTTP_ENDPOINT); + httpEndpointObj = httpEndpointObj != null ? httpEndpointObj : span.unsafeGetTag(HTTP_ROUTE); httpEndpoint = httpEndpointObj != null ? httpEndpointObj.toString() : null; } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java index 81de46c9113..3c174c895c8 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java @@ -17,6 +17,12 @@ public static MetricsAggregator createMetricsAggregator( // the same ClientStatsAggregator span selection + DDSketch aggregation, differing only in // the injected MetricWriter. if (config.isTracesSpanMetricsEnabled()) { + if (config.isTracerMetricsEnabled()) { + log.warn( + "Both OTLP trace span metrics and native tracer metrics are enabled; " + + "using OTLP export and ignoring native tracer metrics (the two are mutually " + + "exclusive)."); + } log.debug("OTLP trace span metrics enabled"); return new ClientStatsAggregator( config, sharedCommunicationObjects, healthMetrics, new OtlpStatsMetricWriter(config)); diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java index 12e6d74b719..df0e6f7fd1d 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java @@ -7,6 +7,7 @@ import static datadog.trace.common.writer.ddagent.Prioritization.FAST_LANE; import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.communication.ddagent.DroppingPolicy; import datadog.metrics.api.Monitoring; import datadog.trace.api.Config; import datadog.trace.api.ProtocolVersion; @@ -48,6 +49,7 @@ public static class DDAgentWriterBuilder { private DDAgentApi agentApi; private Prioritization prioritization; private DDAgentFeaturesDiscovery featureDiscovery; + private DroppingPolicy droppingPolicy; private SingleSpanSampler singleSpanSampler; public DDAgentWriterBuilder agentApi(DDAgentApi agentApi) { @@ -125,6 +127,11 @@ public DDAgentWriterBuilder featureDiscovery(DDAgentFeaturesDiscovery featureDis return this; } + public DDAgentWriterBuilder droppingPolicy(DroppingPolicy droppingPolicy) { + this.droppingPolicy = droppingPolicy; + return this; + } + public DDAgentWriterBuilder flushTimeout(int flushTimeout, TimeUnit flushTimeoutUnit) { this.flushTimeout = flushTimeout; this.flushTimeoutUnit = flushTimeoutUnit; @@ -170,7 +177,10 @@ public DDAgentWriter build() { traceBufferSize, healthMetrics, dispatcher, - featureDiscovery, + droppingPolicy != null + ? droppingPolicy + : featureDiscovery, // custom dropping policy for OTLP but backup to feature + // discovery null == prioritization ? FAST_LANE : prioritization, flushIntervalMilliseconds, TimeUnit.MILLISECONDS, diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java index fc8a1ef00cc..14de2610e35 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java @@ -193,10 +193,18 @@ public static Writer createWriter( ddAgentApi.addResponseListener((RemoteResponseListener) sampler); } + // Drop p0 (sampled-out) traces when client-side stats are being computed -- either via the + // native agent-stats path (featuresDiscovery) or the OTLP trace metrics path -- so the + // now-redundant traces aren't shipped. + final boolean otlpSpanMetricsEnabled = config.isTracesSpanMetricsEnabled(); + final DroppingPolicy droppingPolicy = + () -> otlpSpanMetricsEnabled || featuresDiscovery.active(); + DDAgentWriter.DDAgentWriterBuilder builder = DDAgentWriter.builder() .agentApi(ddAgentApi) .featureDiscovery(featuresDiscovery) + .droppingPolicy(droppingPolicy) .prioritization(prioritization) .healthMetrics(healthMetrics) .monitoring(commObjects.monitoring) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java index d65b2f77b80..b0d0f4bc240 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java @@ -109,7 +109,8 @@ public Response sendSerializedTraces(final Payload payload) { .addHeader(DATADOG_DROPPED_SPAN_COUNT, Long.toString(payload.droppedSpans())) .addHeader( DATADOG_CLIENT_COMPUTED_STATS, - (metricsEnabled && featuresDiscovery.supportsMetrics()) + Config.get().isTracesSpanMetricsEnabled() + || (metricsEnabled && featuresDiscovery.supportsMetrics()) // Disabling the computation agent-side of the APM trace metrics by // pretending it was already done by the library || !Config.get().isApmTracingEnabled() From 7c8bfeb6f7aa7f29e334e4d3be0334634821de44 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Tue, 23 Jun 2026 15:21:22 -0400 Subject: [PATCH 3/7] remove unnecessary code --- .../java/datadog/trace/common/metrics/ClientStatsAggregator.java | 1 - 1 file changed, 1 deletion(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index f8da68407dc..2b262af4946 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -351,7 +351,6 @@ private boolean publish(CoreSpan span, boolean isTopLevel, PeerTagSchema peer Object httpMethodObj = span.unsafeGetTag(HTTP_METHOD); httpMethod = httpMethodObj != null ? httpMethodObj.toString() : null; Object httpEndpointObj = span.unsafeGetTag(HTTP_ENDPOINT); - httpEndpointObj = httpEndpointObj != null ? httpEndpointObj : span.unsafeGetTag(HTTP_ROUTE); httpEndpoint = httpEndpointObj != null ? httpEndpointObj.toString() : null; } From d17e466056563b5406490c56505baaff6299d1a6 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Sat, 27 Jun 2026 09:53:38 -0400 Subject: [PATCH 4/7] cleanup --- .../trace/common/metrics/OtlpStatsMetricWriter.java | 7 ------- 1 file changed, 7 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 7915d61bd4d..da1ad8e667b 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 @@ -72,13 +72,6 @@ public final class OtlpStatsMetricWriter implements MetricWriter { @Nullable private final OtlpSender sender; private final boolean otelSemanticsMode; - /** - * Resource attribute blob prepended to every payload. In default mode it carries the {@code - * datadog.runtime_id} and process-tag resource attributes; in OTel-semantics mode it is the plain - * vendor-neutral resource (no {@code datadog.*}). - */ - private final byte[] resourceMessage; - // 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); From cb593817817ebc3c7d3696ceaf3f5a253b2bb992 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 1 Jul 2026 15:37:27 -0400 Subject: [PATCH 5/7] cleanup --- .../common/metrics/ClientStatsAggregator.java | 21 ++++- .../metrics/MetricsAggregatorFactory.java | 12 +++ .../trace/common/metrics/NoOpSink.java | 8 +- .../trace/common/writer/DDAgentWriter.java | 18 ++--- .../trace/common/writer/WriterFactory.java | 5 +- .../common/writer/ddagent/DDAgentApi.java | 8 +- .../trace/common/writer/DDAgentApiTest.groovy | 45 +++++++++++ .../common/writer/WriterFactoryTest.groovy | 45 ----------- .../metrics/MetricsAggregatorFactoryTest.java | 81 +++++-------------- .../otlp/common/OtlpResourceProtoTest.java | 4 - 10 files changed, 109 insertions(+), 138 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index 2b262af4946..d3c0e7ef1ef 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -219,6 +219,22 @@ public ClientStatsAggregator( this.reportingIntervalTimeUnit = timeUnit; } + // ── visible for testing ───────────────────────────────────────────────────── + // Expose the writer-selection outcome and reporting cadence so tests can assert + // the native-vs-OTLP XOR choice without reflecting into private fields. + + boolean isOtlpStatsExportEnabled() { + return otlpStatsExportEnabled; + } + + long reportingInterval() { + return reportingInterval; + } + + TimeUnit reportingIntervalTimeUnit() { + return reportingIntervalTimeUnit; + } + @Override public void start() { sink.register(this); @@ -239,10 +255,7 @@ private boolean statsExportEnabled() { } private boolean isMetricsEnabled() { - // The discovery refresh only helps the native path, which is gated on the agent advertising - // v0.6/stats. The OTLP path uses its own sender and never depends on that capability -- even - // when a Datadog Agent is present (the OTLP endpoint may itself be the agent's OTLP receiver), - // so refreshing agent features here would be pointless for it. + // The discovery refresh only helps the native path. if (!otlpStatsExportEnabled && features.getMetricsEndpoint() == null) { features.discoverIfOutdated(); } diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java index 3c174c895c8..e7d2b72e9b7 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java @@ -23,6 +23,18 @@ public static MetricsAggregator createMetricsAggregator( + "using OTLP export and ignoring native tracer metrics (the two are mutually " + "exclusive)."); } + if (!config.isMetricsOtlpExporterEnabled()) { + // Reachable only via an explicit OTEL_TRACES_SPAN_METRICS_ENABLED=true: the implicit + // default already requires the OTLP metrics exporter. The span-metrics writer is not + // gated by metrics.otel.exporter -- it builds its sender from the otlp.metrics.* transport + // settings -- so metrics still leave over OTLP even though the general OTel metrics + // exporter is not OTLP. Warn so this is not a silent surprise. + log.warn( + "OTLP trace span metrics are enabled but the OTLP metrics exporter is not " + + "(metrics.otel.exporter is not 'otlp'); span metrics will still be exported over " + + "OTLP using the otlp.metrics.* transport settings. Set metrics.otel.exporter=otlp " + + "to make this explicit, or disable traces.span.metrics.enabled to suppress them."); + } log.debug("OTLP trace span metrics enabled"); return new ClientStatsAggregator( config, sharedCommunicationObjects, healthMetrics, new OtlpStatsMetricWriter(config)); diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java index f624f5f403c..75236791324 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java @@ -2,13 +2,7 @@ import java.nio.ByteBuffer; -/** - * A {@link Sink} that discards everything. Used by the OTLP trace-metrics export path, where {@link - * OtlpStatsMetricWriter} sends payloads directly via its own OTLP sender in {@code finishBucket()} - * rather than through the aggregator's {@code Sink}. {@link ClientStatsAggregator} still - * requires a {@code Sink} for {@code register()}/backpressure wiring, so this satisfies that - * contract without performing any I/O. - */ +/** A {@link Sink} that discards everything. */ public final class NoOpSink implements Sink { public static final NoOpSink INSTANCE = new NoOpSink(); diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java index df0e6f7fd1d..dc4da3545bd 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java @@ -40,7 +40,7 @@ public static class DDAgentWriterBuilder { int flushIntervalMilliseconds = 1000; Monitoring monitoring = Monitoring.DISABLED; ProtocolVersion protocolVersion = Config.get().getProtocolVersion(); - boolean metricsReportingEnabled = Config.get().isTracerMetricsEnabled(); + boolean nativeMetricsReportingEnabled = Config.get().isTracerMetricsEnabled(); boolean metricsIgnoreAgentVersion = Config.get().isTracerMetricsIgnoreAgentVersion(); private int flushTimeout = 1; private TimeUnit flushTimeoutUnit = TimeUnit.SECONDS; @@ -112,8 +112,9 @@ public DDAgentWriterBuilder traceAgentProtocolVersion(ProtocolVersion protocolVe return this; } - public DDAgentWriterBuilder metricsReportingEnabled(boolean metricsReportingEnabled) { - this.metricsReportingEnabled = metricsReportingEnabled; + public DDAgentWriterBuilder nativeMetricsReportingEnabled( + boolean nativeMetricsReportingEnabled) { + this.nativeMetricsReportingEnabled = nativeMetricsReportingEnabled; return this; } @@ -161,12 +162,13 @@ public DDAgentWriter build() { monitoring, agentUrl, protocolVersion, - metricsReportingEnabled, + nativeMetricsReportingEnabled, metricsIgnoreAgentVersion); } if (null == agentApi) { agentApi = - new DDAgentApi(client, agentUrl, featureDiscovery, monitoring, metricsReportingEnabled); + new DDAgentApi( + client, agentUrl, featureDiscovery, monitoring, nativeMetricsReportingEnabled); } final DDAgentMapperDiscovery mapperDiscovery = new DDAgentMapperDiscovery(featureDiscovery); @@ -177,10 +179,8 @@ public DDAgentWriter build() { traceBufferSize, healthMetrics, dispatcher, - droppingPolicy != null - ? droppingPolicy - : featureDiscovery, // custom dropping policy for OTLP but backup to feature - // discovery + // allow custom dropping policy for OTLP but backup to feature discovery + droppingPolicy != null ? droppingPolicy : featureDiscovery, null == prioritization ? FAST_LANE : prioritization, flushIntervalMilliseconds, TimeUnit.MILLISECONDS, diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java index 14de2610e35..e91b183216e 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/WriterFactory.java @@ -187,15 +187,14 @@ public static Writer createWriter( commObjects.agentUrl, featuresDiscovery, commObjects.monitoring, - config.isTracerMetricsEnabled() || config.isTracesSpanMetricsEnabled()); + config.isTracerMetricsEnabled()); if (sampler instanceof RemoteResponseListener) { ddAgentApi.addResponseListener((RemoteResponseListener) sampler); } // Drop p0 (sampled-out) traces when client-side stats are being computed -- either via the - // native agent-stats path (featuresDiscovery) or the OTLP trace metrics path -- so the - // now-redundant traces aren't shipped. + // native agent-stats path (featuresDiscovery) or the OTLP trace metrics path final boolean otlpSpanMetricsEnabled = config.isTracesSpanMetricsEnabled(); final DroppingPolicy droppingPolicy = () -> otlpSpanMetricsEnabled || featuresDiscovery.active(); diff --git a/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java index b0d0f4bc240..7adb34adbd6 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/writer/ddagent/DDAgentApi.java @@ -43,7 +43,7 @@ public class DDAgentApi extends RemoteApi { private static final String DATADOG_AGENT_STATE = "Datadog-Agent-State"; private final List responseListeners = new ArrayList<>(); - private final boolean metricsEnabled; + private final boolean nativeMetricsEnabled; private final Recording sendPayloadTimer; private final Counter agentErrorCounter; @@ -67,14 +67,14 @@ public DDAgentApi( HttpUrl agentUrl, DDAgentFeaturesDiscovery featuresDiscovery, Monitoring monitoring, - boolean metricsEnabled) { + boolean nativeMetricsEnabled) { super(false); this.featuresDiscovery = featuresDiscovery; this.agentUrl = agentUrl; this.httpClient = client; this.sendPayloadTimer = monitoring.newTimer("trace.agent.send.time"); this.agentErrorCounter = monitoring.newCounter("trace.agent.error.counter"); - this.metricsEnabled = metricsEnabled; + this.nativeMetricsEnabled = nativeMetricsEnabled; this.headers = new HashMap<>(); this.headers.put(DATADOG_CLIENT_COMPUTED_TOP_LEVEL, "true"); @@ -110,7 +110,7 @@ public Response sendSerializedTraces(final Payload payload) { .addHeader( DATADOG_CLIENT_COMPUTED_STATS, Config.get().isTracesSpanMetricsEnabled() - || (metricsEnabled && featuresDiscovery.supportsMetrics()) + || (nativeMetricsEnabled && featuresDiscovery.supportsMetrics()) // Disabling the computation agent-side of the APM trace metrics by // pretending it was already done by the library || !Config.get().isApmTracingEnabled() diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentApiTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentApiTest.groovy index e4cc7d88fd3..2ee83f1e5e3 100644 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentApiTest.groovy +++ b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/DDAgentApiTest.groovy @@ -18,6 +18,7 @@ import datadog.metrics.impl.MonitoringImpl import datadog.trace.api.Config import datadog.trace.api.ProcessTags import datadog.trace.api.ProtocolVersion +import datadog.trace.api.config.OtlpConfig import datadog.trace.bootstrap.instrumentation.api.InstrumentationTags import datadog.trace.common.sampling.RateByServiceTraceSampler import datadog.trace.common.writer.ddagent.DDAgentApi @@ -454,6 +455,50 @@ class DDAgentApiTest extends DDCoreSpecification { return maps } + def "Datadog-Client-Computed-Stats header set when either stats pipeline is enabled (otlpSpanMetrics=#otlpSpanMetrics, nativeMetrics=#nativeMetrics)"() { + setup: + injectSysConfig(OtlpConfig.TRACES_SPAN_METRICS_ENABLED, "$otlpSpanMetrics") + + def agent = httpServer { + handlers { + put("v0.4/traces") { + response.status(200).send() + } + } + } + def agentUrl = HttpUrl.get(agent.address.toString()) + + // Mock feature discovery so the native-stats pipeline signal can be controlled independently of + // what the (embedded) agent advertises. supportsMetrics() reflects agent-side client stats support. + def discovery = Mock(DDAgentFeaturesDiscovery) + discovery.getTraceEndpoint() >> "v0.4/traces" + discovery.supportsMetrics() >> nativeMetrics + discovery.state() >> null + + def client = OkHttpUtils.buildHttpClient(agentUrl, 1000) + // nativeMetricsEnabled is the constructor flag WriterFactory sets from Config.isTracerMetricsEnabled(). + def api = new DDAgentApi(client, agentUrl, discovery, monitoring, nativeMetrics) + def payload = prepareTraces("v0.4/traces", []) + + when: + // Named clientResponse (not response) to avoid shadowing the httpServer handler closure's `response`. + def clientResponse = api.sendSerializedTraces(payload) + + then: + clientResponse.success() + (agent.lastRequest.headers.get("Datadog-Client-Computed-Stats") == "true") == expectedComputesStats + + cleanup: + agent.close() + + where: + otlpSpanMetrics | nativeMetrics | expectedComputesStats + true | false | true // gap case: OTLP span metrics on, native stats off + false | false | false // neither pipeline computes stats + false | true | true // native stats on (regression guard) + true | true | true // both on + } + Payload prepareTraces(String agentVersion, List> traces) { Traces traceCapture = new Traces() def packer = new MsgPackWriter(new FlushingBuffer(1 << 20, traceCapture)) diff --git a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy index 59cc60c3820..d20bd475cac 100644 --- a/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy +++ b/dd-trace-core/src/test/groovy/datadog/trace/common/writer/WriterFactoryTest.groovy @@ -207,51 +207,6 @@ class WriterFactoryTest extends DDSpecification { OtlpConfig.Protocol.GRPC | OtlpConfig.Compression.NONE | "http://otel-collector:4317" | OtlpGrpcSender | "http://otel-collector:4317/opentelemetry.proto.collector.trace.v1.TraceService/Export" | false } - def "DDAgentApi marks client-computed stats when either stats pipeline is enabled (spanMetrics=#spanMetrics, nativeStats=#nativeStats)"() { - setup: - // metricsEnabled gates the Datadog-Client-Computed-Stats header on the native trace transport. - // It must be true whenever the SDK computes stats by EITHER pipeline: native msgpack stats - // (isTracerMetricsEnabled) or OTLP span metrics (isTracesSpanMetricsEnabled). The OTLP-stats - // case arises when OTEL_TRACES_SPAN_METRICS_ENABLED is set while traces still export natively; - // without the header the Agent would recompute stats from spans already exported via OTLP. - def config = Mock(Config) - config.apiKey >> "my-api-key" - config.agentUrl >> "http://my-agent.url" - config.getEnumValue(PRIORITIZATION_TYPE, _, _) >> Prioritization.FAST_LANE - config.tracerMetricsEnabled >> nativeStats - config.tracesSpanMetricsEnabled >> spanMetrics - - def mockCall = Mock(Call) - def mockHttpClient = Mock(OkHttpClient) - // advertise only v0.4 (no EVP proxy) so "DDAgentWriter" resolves to a real DDAgentWriter - mockCall.execute() >> { buildHttpResponse(false, false, HttpUrl.parse(config.agentUrl + "/info")) } - mockHttpClient.newCall(_ as Request) >> mockCall - - def sharedComm = new SharedCommunicationObjects() - sharedComm.agentHttpClient = mockHttpClient - sharedComm.agentUrl = HttpUrl.parse(config.agentUrl) - sharedComm.createRemaining(config) - - def sampler = Mock(Sampler) - - when: - def writer = WriterFactory.createWriter(config, sharedComm, sampler, null, HealthMetrics.NO_OP, "DDAgentWriter") - def api = ((RemoteWriter) writer).apis.find { it instanceof DDAgentApi } - def metricsEnabledField = DDAgentApi.getDeclaredField("metricsEnabled") - metricsEnabledField.setAccessible(true) - - then: - writer instanceof DDAgentWriter - metricsEnabledField.getBoolean(api) == expectedComputesStats - - where: - spanMetrics | nativeStats | expectedComputesStats - true | false | true // gap case: OTLP span metrics on, native stats off - false | false | false // neither pipeline computes stats - false | true | true // native stats on (regression guard) - true | true | true // both on - } - Response buildHttpResponse(boolean hasEvpProxy, boolean evpProxySupportsCompression, HttpUrl agentUrl) { def endpoints = [] if (hasEvpProxy && evpProxySupportsCompression) { diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java index 7453a9984c7..9fb3407cf42 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java @@ -3,10 +3,12 @@ import static datadog.trace.api.config.GeneralConfig.TRACE_STATS_COMPUTATION_ENABLED; import static datadog.trace.api.config.OtlpConfig.OTLP_METRICS_PROTOCOL; import static datadog.trace.api.config.OtlpConfig.TRACES_SPAN_METRICS_ENABLED; -import static datadog.trace.api.config.OtlpConfig.TRACE_OTEL_EXPORTER; import static java.util.concurrent.TimeUnit.MILLISECONDS; +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.assertInstanceOf; +import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; @@ -15,16 +17,14 @@ import datadog.communication.ddagent.SharedCommunicationObjects; import datadog.trace.api.Config; import datadog.trace.core.monitor.HealthMetrics; -import java.lang.reflect.Field; import java.util.Properties; -import java.util.concurrent.TimeUnit; import okhttp3.HttpUrl; import org.junit.jupiter.api.Test; /** * Tests the native-vs-OTLP XOR writer selection in {@link MetricsAggregatorFactory}. The selected - * {@link MetricWriter} is not exposed publicly, so it is read via reflection through {@code - * ClientStatsAggregator.aggregator -> Aggregator.writer}. + * writer is not exposed directly, but {@link ClientStatsAggregator} publishes the selection + * outcome via {@code isOtlpStatsExportEnabled()} plus the reporting cadence getters. */ class MetricsAggregatorFactoryTest { @@ -55,7 +55,7 @@ void whenAllMetricsDisabledNoOpAggregatorCreated() { } @Test - void whenNativeTracerMetricsEnabledSerializingWriterSelected() throws Exception { + void whenNativeTracerMetricsEnabledSerializingWriterSelected() { // tracer metrics default to enabled; OTLP span metrics default off (no OTLP trace export). Config config = Config.get(props()); @@ -63,40 +63,16 @@ void whenNativeTracerMetricsEnabledSerializingWriterSelected() throws Exception MetricsAggregatorFactory.createMetricsAggregator( config, sharedCommunicationObjects(), HealthMetrics.NO_OP); - assertInstanceOf(ClientStatsAggregator.class, aggregator); - assertInstanceOf(SerializingMetricWriter.class, writerOf(aggregator)); + ClientStatsAggregator conflating = + assertInstanceOf(ClientStatsAggregator.class, aggregator); + assertFalse(conflating.isOtlpStatsExportEnabled()); + // native path uses a hardcoded 10s cadence, not trace.stats.interval. + assertEquals(10, conflating.reportingInterval()); + assertEquals(SECONDS, conflating.reportingIntervalTimeUnit()); } @Test - void whenOtlpSpanMetricsEnabledOtlpWriterSelectedWithConfiguredInterval() throws Exception { - // OTEL_TRACES_EXPORTER=otlp satisfies FR16 (traces not routed through an Agent). http/json has - // no protobuf-free encoder, so the writer's sender is null -- keeps construction lightweight - // (no - // real OTLP sender / network) while still exercising writer selection. - Config config = - Config.get( - props( - TRACES_SPAN_METRICS_ENABLED, "true", - TRACE_OTEL_EXPORTER, "otlp", - OTLP_METRICS_PROTOCOL, "http/json")); - - MetricsAggregator aggregator = - MetricsAggregatorFactory.createMetricsAggregator( - config, sharedCommunicationObjects(), HealthMetrics.NO_OP); - - assertInstanceOf(ClientStatsAggregator.class, aggregator); - assertInstanceOf(OtlpStatsMetricWriter.class, writerOf(aggregator)); - // reporting interval comes from getTraceStatsInterval() (ms), default 10s. - assertEquals(config.getTraceStatsInterval(), reportingIntervalOf(aggregator)); - assertEquals(MILLISECONDS, reportingIntervalUnitOf(aggregator)); - } - - @Test - void explicitSpanMetricsEnabledSelectsOtlpWriterEvenWithoutOtlpTraceExport() throws Exception { - // An explicit OTEL_TRACES_SPAN_METRICS_ENABLED=true is honored verbatim regardless of trace - // exporter (matching the dd-trace-py / dd-trace-go reference impls). Double-counting if these - // OTLP spans reach an Agent is handled by the FR15 _dd.stats_computed marker, not by disabling - // here. http/json keeps the writer's sender null (no network at construction). + void whenOtlpTraceMetricsEnabledOtlpStatsMetricWriterSelected() { Config config = Config.get(props(TRACES_SPAN_METRICS_ENABLED, "true", OTLP_METRICS_PROTOCOL, "http/json")); @@ -104,30 +80,11 @@ void explicitSpanMetricsEnabledSelectsOtlpWriterEvenWithoutOtlpTraceExport() thr MetricsAggregatorFactory.createMetricsAggregator( config, sharedCommunicationObjects(), HealthMetrics.NO_OP); - assertInstanceOf(ClientStatsAggregator.class, aggregator); - assertInstanceOf(OtlpStatsMetricWriter.class, writerOf(aggregator)); - } - - // ── reflection helpers ───────────────────────────────────────────────────── - - private static MetricWriter writerOf(MetricsAggregator aggregator) throws Exception { - Field aggregatorField = ClientStatsAggregator.class.getDeclaredField("aggregator"); - aggregatorField.setAccessible(true); - Object inner = aggregatorField.get(aggregator); - Field writerField = Aggregator.class.getDeclaredField("writer"); - writerField.setAccessible(true); - return (MetricWriter) writerField.get(inner); - } - - private static long reportingIntervalOf(MetricsAggregator aggregator) throws Exception { - Field f = ClientStatsAggregator.class.getDeclaredField("reportingInterval"); - f.setAccessible(true); - return f.getLong(aggregator); - } - - private static TimeUnit reportingIntervalUnitOf(MetricsAggregator aggregator) throws Exception { - Field f = ClientStatsAggregator.class.getDeclaredField("reportingIntervalTimeUnit"); - f.setAccessible(true); - return (TimeUnit) f.get(aggregator); + ClientStatsAggregator conflating = + assertInstanceOf(ClientStatsAggregator.class, aggregator); + assertTrue(conflating.isOtlpStatsExportEnabled()); + // OTLP path sources the cadence from trace.stats.interval (ms), default 10s. + assertEquals(config.getTraceStatsInterval(), conflating.reportingInterval()); + assertEquals(MILLISECONDS, conflating.reportingIntervalTimeUnit()); } } 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 5978398d0ec..1a1b958e4f1 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 @@ -179,10 +179,6 @@ void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException { assertFalse(plain.containsKey("datadog.runtime_id"), "plain variant omits datadog.runtime_id"); } - /** - * FR15: the {@code includeStatsComputed} variant adds the {@code _dd.stats_computed=true} marker - * (used on the OTLP trace payload so a downstream Agent skips recompute); the default omits it. - */ @Test void statsComputedVariantCarriesMarker() throws IOException { Config config = Config.get(props(SERVICE_NAME, "my-service")); From 51c7dce2abbbf840e5f1aae9ba2e9d7ab0f77c3c Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Wed, 1 Jul 2026 15:39:36 -0400 Subject: [PATCH 6/7] cleanup pt 2 --- .../trace/common/metrics/MetricsAggregatorFactory.java | 5 ----- 1 file changed, 5 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java index e7d2b72e9b7..0cb948cb1d5 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/MetricsAggregatorFactory.java @@ -24,11 +24,6 @@ public static MetricsAggregator createMetricsAggregator( + "exclusive)."); } if (!config.isMetricsOtlpExporterEnabled()) { - // Reachable only via an explicit OTEL_TRACES_SPAN_METRICS_ENABLED=true: the implicit - // default already requires the OTLP metrics exporter. The span-metrics writer is not - // gated by metrics.otel.exporter -- it builds its sender from the otlp.metrics.* transport - // settings -- so metrics still leave over OTLP even though the general OTel metrics - // exporter is not OTLP. Warn so this is not a silent surprise. log.warn( "OTLP trace span metrics are enabled but the OTLP metrics exporter is not " + "(metrics.otel.exporter is not 'otlp'); span metrics will still be exported over " From 3d2a08873b5d302f68be256bebd1b168628a6cb6 Mon Sep 17 00:00:00 2001 From: Matthew Li Date: Thu, 2 Jul 2026 19:55:53 -0500 Subject: [PATCH 7/7] cleanup pt 2 --- .../trace/common/metrics/ClientStatsAggregator.java | 1 - .../trace/common/metrics/OtlpStatsMetricWriter.java | 7 +++++++ .../common/metrics/MetricsAggregatorFactoryTest.java | 10 ++++------ 3 files changed, 11 insertions(+), 7 deletions(-) diff --git a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java index d3c0e7ef1ef..b3797261a3d 100644 --- a/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/ClientStatsAggregator.java @@ -4,7 +4,6 @@ import static datadog.trace.api.DDSpanTypes.RPC; import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ENDPOINT; import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_METHOD; -import static datadog.trace.bootstrap.instrumentation.api.Tags.HTTP_ROUTE; import static datadog.trace.common.metrics.AggregateEntry.ERROR_TAG; import static datadog.trace.common.metrics.AggregateEntry.TOP_LEVEL_TAG; import static datadog.trace.common.metrics.SignalItem.ClearSignal.CLEAR; 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 da1ad8e667b..7915d61bd4d 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 @@ -72,6 +72,13 @@ public final class OtlpStatsMetricWriter implements MetricWriter { @Nullable private final OtlpSender sender; private final boolean otelSemanticsMode; + /** + * Resource attribute blob prepended to every payload. In default mode it carries the {@code + * datadog.runtime_id} and process-tag resource attributes; in OTel-semantics mode it is the plain + * vendor-neutral resource (no {@code datadog.*}). + */ + private final byte[] resourceMessage; + // 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); diff --git a/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java index 9fb3407cf42..c77559fa5c2 100644 --- a/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java @@ -23,8 +23,8 @@ /** * Tests the native-vs-OTLP XOR writer selection in {@link MetricsAggregatorFactory}. The selected - * writer is not exposed directly, but {@link ClientStatsAggregator} publishes the selection - * outcome via {@code isOtlpStatsExportEnabled()} plus the reporting cadence getters. + * writer is not exposed directly, but {@link ClientStatsAggregator} publishes the selection outcome + * via {@code isOtlpStatsExportEnabled()} plus the reporting cadence getters. */ class MetricsAggregatorFactoryTest { @@ -63,8 +63,7 @@ void whenNativeTracerMetricsEnabledSerializingWriterSelected() { MetricsAggregatorFactory.createMetricsAggregator( config, sharedCommunicationObjects(), HealthMetrics.NO_OP); - ClientStatsAggregator conflating = - assertInstanceOf(ClientStatsAggregator.class, aggregator); + ClientStatsAggregator conflating = assertInstanceOf(ClientStatsAggregator.class, aggregator); assertFalse(conflating.isOtlpStatsExportEnabled()); // native path uses a hardcoded 10s cadence, not trace.stats.interval. assertEquals(10, conflating.reportingInterval()); @@ -80,8 +79,7 @@ void whenOtlpTraceMetricsEnabledOtlpStatsMetricWriterSelected() { MetricsAggregatorFactory.createMetricsAggregator( config, sharedCommunicationObjects(), HealthMetrics.NO_OP); - ClientStatsAggregator conflating = - assertInstanceOf(ClientStatsAggregator.class, aggregator); + ClientStatsAggregator conflating = assertInstanceOf(ClientStatsAggregator.class, aggregator); assertTrue(conflating.isOtlpStatsExportEnabled()); // OTLP path sources the cadence from trace.stats.interval (ms), default 10s. assertEquals(config.getTraceStatsInterval(), conflating.reportingInterval());