Wire OTLP Trace Metrics Writers into Metrics Aggregators

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,

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<String> 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<Boolean> forceReport() {
   public boolean publish(List<? extends CoreSpan<?>> 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

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);

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) {}
+}

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,

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)

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<RemoteResponseListener> 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()

dd-trace-core/src/main/java/datadog/trace/core/otlp/common/OtlpResourceProto.java

@@ -36,22 +36,33 @@ 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
    * {@code datadog.}). Used by the default-mode SDK trace-metrics export; omitted in OTel-semantics
    * 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];

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);