DataDog · mhlidd · Jul 17, 2026 · Jun 17, 2026 · Jun 30, 2026 · Jun 18, 2026
@@ -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,

@@ -31,7 +31,7 @@ public final class OtlpConfig {
   public static final String OTLP_METRICS_TEMPORALITY_PREFERENCE =
       "otlp.metrics.temporality.preference";
 
-  public static final String TRACES_SPAN_METRICS_ENABLED = "traces.span.metrics.enabled";
+  public static final String OTEL_TRACES_SPAN_METRICS_ENABLED = "otel.traces.span.metrics.enabled";
 
   public static final String TRACE_OTEL_ENABLED = "trace.otel.enabled";
   public static final String TRACE_OTEL_EXPORTER = "trace.otel.exporter";

@@ -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;
@@ -12,6 +13,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;
@@ -26,6 +28,7 @@
 import datadog.trace.core.DDTraceCoreInfo;
 import datadog.trace.core.SpanKindFilter;
 import datadog.trace.core.monitor.HealthMetrics;
+import datadog.trace.core.otlp.metrics.OtlpStatsMetricWriter;
 import datadog.trace.util.AgentTaskScheduler;
 import java.util.Collections;
 import java.util.List;
@@ -62,6 +65,15 @@ public final class ClientStatsAggregator implements MetricsAggregator, EventList
   private static final SpanKindFilter INTERNAL_KIND =
       SpanKindFilter.builder().includeInternal().build();
 
+  // gRPC status-code source tags, probed in priority order on the OTLP export path
+  private static final String[] GRPC_STATUS_CODE_KEYS = {
+    InstrumentationTags.GRPC_STATUS_CODE, // "rpc.grpc.status_code"
+    "grpc.code",
+    "rpc.grpc.status.code",
+    "grpc.status.code",
+    "rpc.response.status_code",
+  };
+
   private final Set<String> ignoredResources;
   private final Thread thread;
   private final MessagePassingQueue<InboxItem> inbox;
@@ -72,6 +84,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 +127,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. 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 +210,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 +229,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 +260,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 +325,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
@@ -315,12 +374,21 @@ 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);
+      // OTLP path falls back to http.route (mirrors libdatadog). The native v0.6 path keeps its
+      // http.endpoint-only lookup so this doesn't change its aggregation key / wire output.
+      if (otlpStatsExportEnabled && httpEndpointObj == null) {
+        httpEndpointObj = span.unsafeGetTag(HTTP_ROUTE);
+      }
       httpEndpoint = httpEndpointObj != null ? httpEndpointObj.toString() : null;
     }
 
     CharSequence spanType = span.getType();
     String grpcStatusCode = null;
-    if (spanType != null && RPC.contentEquals(spanType)) {
+    if (otlpStatsExportEnabled) {
+      // OTLP path: probe every known gRPC status-code convention, no span-type gate, so a span
+      // typed "grpc" (or carrying an OTel-style key) still surfaces rpc.response.status_code.
+      grpcStatusCode = firstTag(span, GRPC_STATUS_CODE_KEYS);
+    } else if (spanType != null && RPC.contentEquals(spanType)) {
       Object grpcStatusObj = span.unsafeGetTag(InstrumentationTags.GRPC_STATUS_CODE);
       grpcStatusCode = grpcStatusObj != null ? grpcStatusObj.toString() : null;
     }
@@ -367,6 +435,17 @@ private boolean publish(CoreSpan<?> span, boolean isTopLevel, PeerTagSchema peer
     return error;
   }
 
+  /** Returns the first non-null span tag among {@code keys}, in order, or {@code null} if none. */
+  private static String firstTag(CoreSpan<?> span, String[] keys) {
+    for (String key : keys) {
+      Object value = span.unsafeGetTag(key);
+      if (value != null) {
+        return value.toString();
+      }
+    }
+    return null;
+  }
+
   /**
    * One-time producer-side bootstrap of {@link #cachedPeerTagSchema}. Synchronized double-check
    * guards against two producers racing on the very first publish; after this returns, {@code

@@ -1,8 +1,12 @@
 package datadog.trace.common.metrics;
 
+import static datadog.trace.api.config.OtlpConfig.METRICS_OTEL_EXPORTER;
+import static datadog.trace.api.config.OtlpConfig.OTEL_TRACES_SPAN_METRICS_ENABLED;
+
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.trace.api.Config;
 import datadog.trace.core.monitor.HealthMetrics;
+import datadog.trace.core.otlp.metrics.OtlpStatsMetricWriter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -13,6 +17,33 @@ 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.isOtelTracesSpanMetricsEnabled()) {
+      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 "
+                + OTEL_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);

@@ -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) {}
+}
@@ -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; otherwise fall back to feature discovery
+              droppingPolicy != null ? droppingPolicy : featureDiscovery,
               null == prioritization ? FAST_LANE : prioritization,
               flushIntervalMilliseconds,
               TimeUnit.MILLISECONDS,

@@ -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.isOtelTracesSpanMetricsEnabled();
+      final DroppingPolicy droppingPolicy =
+          () -> otlpSpanMetricsEnabled || featuresDiscovery.active();
+
       DDAgentWriter.DDAgentWriterBuilder builder =
           DDAgentWriter.builder()
               .agentApi(ddAgentApi)
               .featureDiscovery(featuresDiscovery)
+              .droppingPolicy(droppingPolicy)
               .prioritization(prioritization)
               .healthMetrics(healthMetrics)
               .monitoring(commObjects.monitoring)

@@ -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().isOtelTracesSpanMetricsEnabled()
+                          || (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()