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 3a3ded31ff8..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 @@ -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; @@ -72,6 +73,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. @@ -114,6 +116,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, + true); + } + ClientStatsAggregator( WellKnownTags wellKnownTags, Set ignoredResources, @@ -173,6 +199,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; @@ -191,6 +218,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); @@ -206,11 +249,16 @@ 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. + if (!otlpStatsExportEnabled && features.getMetricsEndpoint() == null) { features.discoverIfOutdated(); } - return features.supportsMetrics(); + return statsExportEnabled(); } @Override @@ -266,7 +314,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 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..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 @@ -13,6 +13,27 @@ 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()) { + 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)."); + } + if (!config.isMetricsOtlpExporterEnabled()) { + 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)); + } 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..75236791324 --- /dev/null +++ b/dd-trace-core/src/main/java/datadog/trace/common/metrics/NoOpSink.java @@ -0,0 +1,17 @@ +package datadog.trace.common.metrics; + +import java.nio.ByteBuffer; + +/** A {@link Sink} that discards everything. */ +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/DDAgentWriter.java b/dd-trace-core/src/main/java/datadog/trace/common/writer/DDAgentWriter.java index 12e6d74b719..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 @@ -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; @@ -39,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; @@ -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) { @@ -110,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; } @@ -125,6 +128,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; @@ -154,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); @@ -170,7 +179,8 @@ public DDAgentWriter build() { traceBufferSize, healthMetrics, dispatcher, - featureDiscovery, + // 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 3f38d45881f..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 @@ -193,10 +193,17 @@ 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 + 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..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"); @@ -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() + || (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/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/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/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..c77559fa5c2 --- /dev/null +++ b/dd-trace-core/src/test/java/datadog/trace/common/metrics/MetricsAggregatorFactoryTest.java @@ -0,0 +1,88 @@ +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 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; + +import datadog.communication.ddagent.DDAgentFeaturesDiscovery; +import datadog.communication.ddagent.SharedCommunicationObjects; +import datadog.trace.api.Config; +import datadog.trace.core.monitor.HealthMetrics; +import java.util.Properties; +import okhttp3.HttpUrl; +import org.junit.jupiter.api.Test; + +/** + * 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. + */ +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() { + // 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); + + 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 whenOtlpTraceMetricsEnabledOtlpStatsMetricWriterSelected() { + Config config = + Config.get(props(TRACES_SPAN_METRICS_ENABLED, "true", OTLP_METRICS_PROTOCOL, "http/json")); + + MetricsAggregator aggregator = + MetricsAggregatorFactory.createMetricsAggregator( + config, sharedCommunicationObjects(), HealthMetrics.NO_OP); + + 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 80740dc7e54..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 @@ -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,20 @@ void datadogResourceAttributesVariantCarriesRuntimeId() throws IOException { assertFalse(plain.containsKey("datadog.runtime_id"), "plain variant omits datadog.runtime_id"); } + @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 ─────────────────────────────────────────────────────── /**