feat(tracer): wire OTLP span metrics into tracer flush path

ddtrace/tracer/log.go

@@ -88,7 +88,7 @@ type startupInfo struct {
 // checkEndpoint tries to connect to the URL specified by endpoint.
 // If the endpoint is not reachable, checkEndpoint returns an error
 // explaining why.
-func checkEndpoint(c *http.Client, endpoint string, protocol float64) error {
+func checkEndpoint(c *http.Client, endpoint string, protocol float64, extraHeaders map[string]string) error {
 	b := []byte{0x90} // empty array
 	if protocol == traceProtocolV1 {
 		b = []byte{0x80} // empty map
@@ -99,6 +99,9 @@ func checkEndpoint(c *http.Client, endpoint string, protocol float64) error {
 	}
 	req.Header.Set(traceCountHeader, "0")
 	req.Header.Set("Content-Type", "application/msgpack")
+	for k, v := range extraHeaders {
+		req.Header.Set(k, v)
+	}
 	res, err := c.Do(req)
 	if err != nil {
 		return err
@@ -192,7 +195,11 @@ func logStartup(t *tracer) {
 		info.SampleRateLimit = fmt.Sprintf("%v", limit)
 	}
 	if !t.config.internalConfig.LogToStdout() {
-		if err := checkEndpoint(t.config.httpClient, t.config.ddTransport.endpoint(), t.config.internalConfig.TraceProtocol()); err != nil {
+		var startupHeaders map[string]string
+		if ht, ok := t.config.ddTransport.(*httpTransport); ok {
+			startupHeaders = ht.headers
+		}
+		if err := checkEndpoint(t.config.httpClient, t.config.ddTransport.endpoint(), t.config.internalConfig.TraceProtocol(), startupHeaders); err != nil {
 			info.AgentError = err.Error()
 			log.Warn("DIAGNOSTICS Unable to reach agent intake: %s", err.Error())
 		}

ddtrace/tracer/option.go

@@ -341,9 +341,12 @@ func newConfig(opts ...StartOption) (*config, error) {
 	agentURL := c.internalConfig.AgentURL()
 	af := loadAgentFeatures(agentDisabled, agentURL, c.httpClient)
 	c.agent.store(af)
-	// If the agent doesn't support the v1 protocol, downgrade to v0.4
-	// Also downgrade if CSS is disabled, as v1 is not compatible without CSS.
-	if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats()) {
+	// If the agent doesn't support the v1 protocol, downgrade to v0.4.
+	// Also downgrade if CSS is disabled (v1 requires CSS) or if OTLP span metrics are
+	// enabled: OTLP span metrics use their own concentrator and are not native CSS, so
+	// the trace transport should stay on v0.4 where the test agent and Datadog Agent
+	// can observe the Datadog-Client-Computed-Stats header set by resolveTraceTransport.
+	if c.internalConfig.TraceProtocol() == traceProtocolV1 && (!af.v1ProtocolAvailable || !c.canComputeStats() || c.internalConfig.OTLPSpanMetricsEnabled()) {
 		c.internalConfig.SetTraceProtocol(traceProtocolV04, internalconfig.OriginCalculated)
 		if t, ok := c.ddTransport.(*httpTransport); ok && t.traceURL == agentURL.String()+tracesAPIPathV1 {
 			t.traceURL = agentURL.String() + tracesAPIPath
@@ -442,7 +445,13 @@ func resolveTraceTransport(cfg *internalconfig.Config) (traceURL string, headers
 	if cfg.TraceProtocol() == traceProtocolV1 {
 		traceURL = agentURL + tracesAPIPathV1
 	}
-	return traceURL, datadogHeaders()
+	headers = datadogHeaders()
+	if cfg.OTLPSpanMetricsEnabled() {
+		// Set statically so the header is present on every trace request from startup,
+		// before agent /info polling has completed and CanComputeStats becomes true.
+		headers["Datadog-Client-Computed-Stats"] = "yes"
+	}
+	return traceURL, headers
 }
 
 func newStatsdClient(c *config) (internal.StatsdClient, error) {

ddtrace/tracer/otlp_metrics_wireup_test.go

@@ -0,0 +1,214 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the Apache License Version 2.0.
+// This product includes software developed at Datadog (https://www.datadoghq.com/).
+// Copyright 2025 Datadog, Inc.
+
+package tracer
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/DataDog/datadog-go/v5/statsd"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	tinternal "github.com/DataDog/dd-trace-go/v2/ddtrace/tracer/internal"
+	internalconfig "github.com/DataDog/dd-trace-go/v2/internal/config"
+)
+
+// captureMetricsServer records every POST body it receives.
+type captureMetricsServer struct {
+	mu     sync.Mutex
+	bodies [][]byte
+	*httptest.Server
+}
+
+func newCaptureMetricsServer(t *testing.T) *captureMetricsServer {
+	t.Helper()
+	cs := &captureMetricsServer{}
+	cs.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		b, _ := io.ReadAll(r.Body)
+		cs.mu.Lock()
+		cs.bodies = append(cs.bodies, b)
+		cs.mu.Unlock()
+		w.WriteHeader(http.StatusOK)
+	}))
+	t.Cleanup(cs.Server.Close)
+	return cs
+}
+
+func (cs *captureMetricsServer) receivedBodies() [][]byte {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	out := make([][]byte, len(cs.bodies))
+	copy(out, cs.bodies)
+	return out
+}
+
+// TestOTLPMetricsConcentratorRoutesToExporter verifies that a concentrator wired
+// with an otlpMetricsExporter routes flushed stats to the OTLP endpoint and does
+// not use the agent's native /v0.6/stats path.
+func TestOTLPMetricsConcentratorRoutesToExporter(t *testing.T) {
+	otlpSrv := newCaptureMetricsServer(t)
+
+	dt := newDummyTransport()
+	cfg, err := newTestConfig(withNoopInfoHTTPClient(), func(c *config) {
+		c.ddTransport = dt
+		c.internalConfig.SetEnv("prod", internalconfig.OriginCode)
+	})
+	require.NoError(t, err)
+
+	bucketSize := int64(500_000)
+	c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{})
+	c.otlpExporter = &otlpMetricsExporter{
+		transport: newOTLPTransport(otlpSrv.Server.Client(), otlpSrv.URL+"/v1/metrics", nil),
+		protocol:  "http/json",
+		cfg:       cfg.internalConfig,
+	}
+
+	s := &Span{
+		name:     "http.request",
+		service:  "test-svc",
+		resource: "/api/v1",
+		// 30 seconds in the past ensures its bucket is always before the flush window.
+		start:    time.Now().UnixNano() - int64(30*time.Second),
+		duration: int64(50 * time.Millisecond),
+		metrics:  map[string]float64{keyMeasured: 1},
+	}
+	ss, ok := c.newTracerStatSpan(s, nil)
+	require.True(t, ok)
+
+	// Add the span and flush directly, bypassing goroutine channels to keep the
+	// test deterministic.
+	c.add(ss)
+	c.flushAndSend(time.Now(), withCurrentBucket)
+
+	bodies := otlpSrv.receivedBodies()
+	require.NotEmpty(t, bodies, "OTLP endpoint must receive at least one payload")
+
+	// Verify the payload is valid JSON with a resourceMetrics array.
+	var parsed map[string]any
+	require.NoError(t, json.Unmarshal(bodies[0], &parsed), "OTLP metrics payload must be valid JSON")
+	rm, ok := parsed["resourceMetrics"].([]any)
+	require.True(t, ok, "expected resourceMetrics array in OTLP payload")
+	require.NotEmpty(t, rm)
+
+	// The native /v0.6/stats path must not have been used.
+	assert.Empty(t, dt.Stats(), "native stats path must not be used when otlpExporter is set")
+}
+
+// TestOTLPSpanMetricsHeaderOnNativeTraces verifies that when OTLP span metrics are
+// enabled the Datadog-Client-Computed-Stats: yes header is present on native trace
+// payloads so the agent does not recompute stats.
+func TestOTLPSpanMetricsHeaderOnNativeTraces(t *testing.T) {
+	var headerValue string
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/info" {
+			// No /v0.6/stats endpoint, so CanComputeStats stays false.
+			// The header must still be set via OTLPSpanMetricsEnabled.
+			w.Write([]byte(`{"endpoints":[]}`))
+			return
+		}
+		headerValue = r.Header.Get("Datadog-Client-Computed-Stats")
+	}))
+	defer srv.Close()
+
+	trc, err := newTracer(
+		WithAgentAddr(srv.Listener.Addr().String()),
+		func(c *config) {
+			c.internalConfig.SetOTLPSpanMetricsEnabled(true, internalconfig.OriginCode)
+		},
+	)
+	require.NoError(t, err)
+	setGlobalTracer(trc)
+	defer trc.Stop()
+
+	p, err := encode(getTestTrace(1, 1))
+	require.NoError(t, err)
+	_, err = trc.config.ddTransport.send(p)
+	require.NoError(t, err)
+
+	assert.Equal(t, "yes", headerValue, "Datadog-Client-Computed-Stats must be 'yes' when OTLPSpanMetricsEnabled")
+}
+
+// TestOTLPConcentratorHTTPRouteAttribute verifies that a span carrying the OTel
+// http.route tag (ext.HTTPRoute) produces a data-point with http.route as a
+// first-class attribute (FR06). This exercises the ext.HTTPRoute fallback path
+// in newTracerStatSpan, which populates ClientGroupedStats.HTTPEndpoint so that
+// buildDataPointAttributes emits it — without relying on the peer-tags path.
+func TestOTLPConcentratorHTTPRouteAttribute(t *testing.T) {
+	otlpSrv := newCaptureMetricsServer(t)
+
+	cfg, err := newTestConfig(withNoopInfoHTTPClient())
+	require.NoError(t, err)
+
+	bucketSize := int64(500_000)
+	c := newConcentrator(cfg, bucketSize, &statsd.NoOpClientDirect{})
+	c.otlpExporter = &otlpMetricsExporter{
+		transport: newOTLPTransport(otlpSrv.Server.Client(), otlpSrv.URL+"/v1/metrics", nil),
+		protocol:  "http/json",
+		cfg:       cfg.internalConfig,
+	}
+	c.otlpPeerTags = []string{}
+
+	s := &Span{
+		name:     "web.request",
+		service:  "svc",
+		resource: "web.request",
+		start:    time.Now().UnixNano() - int64(30*time.Second),
+		duration: int64(50 * time.Millisecond),
+		metrics:  map[string]float64{keyMeasured: 1},
+		meta:     tinternal.NewSpanMetaFromMap(map[string]string{"http.route": "/users/{id}"}),
+	}
+	ss, ok := c.newTracerStatSpan(s, nil)
+	require.True(t, ok)
+	c.add(ss)
+	c.flushAndSend(time.Now(), withCurrentBucket)
+
+	bodies := otlpSrv.receivedBodies()
+	require.NotEmpty(t, bodies, "OTLP endpoint must receive a payload")
+
+	body := string(bodies[0])
+	assert.Contains(t, body, `"http.route"`, "http.route key must appear as a data-point attribute")
+	assert.Contains(t, body, `/users/{id}`, "http.route value must appear in the payload")
+}
+
+// TestOTLPTraceWriterStatsComputedResourceAttr verifies that _dd.stats_computed=true
+// is added to the OTLP trace resource when OTLP span metrics are enabled (FR15).
+func TestOTLPTraceWriterStatsComputedResourceAttr(t *testing.T) {
+	t.Run("present-when-enabled", func(t *testing.T) {
+		cfg, err := newTestConfig(func(c *config) {
+			c.internalConfig.SetOTLPSpanMetricsEnabled(true, internalconfig.OriginCode)
+		})
+		require.NoError(t, err)
+
+		w := newOTLPTraceWriter(cfg)
+		var found bool
+		for _, kv := range w.resource.Attributes {
+			if kv.Key == "_dd.stats_computed" {
+				found = true
+				assert.Equal(t, "true", kv.Value.GetStringValue(), "_dd.stats_computed must be string \"true\"")
+			}
+		}
+		assert.True(t, found, "_dd.stats_computed attribute must be present when OTLPSpanMetricsEnabled")
+	})
+
+	t.Run("absent-when-disabled", func(t *testing.T) {
+		cfg, err := newTestConfig(func(c *config) {
+			c.internalConfig.SetOTLPSpanMetricsEnabled(false, internalconfig.OriginCode)
+		})
+		require.NoError(t, err)
+
+		w := newOTLPTraceWriter(cfg)
+		for _, kv := range w.resource.Attributes {
+			assert.NotEqual(t, "_dd.stats_computed", kv.Key,
+				"_dd.stats_computed must not be present when OTLPSpanMetricsEnabled is false")
+		}
+	})
+}

ddtrace/tracer/otlp_transport_test.go

@@ -97,13 +97,13 @@ func TestOTLPTransportSendConnectionError(t *testing.T) {
 }
 
 func TestOTLPTransportConnectionReuse(t *testing.T) {
-	var connCount int64
+	var connCount atomic.Int64
 	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		w.Write([]byte("response body that must be drained"))
 	}))
 	srv.Config.ConnState = func(_ net.Conn, state http.ConnState) {
 		if state == http.StateNew {
-			atomic.AddInt64(&connCount, 1)
+			connCount.Add(1)
 		}
 	}
 	srv.Start()
@@ -113,6 +113,6 @@ func TestOTLPTransportConnectionReuse(t *testing.T) {
 	for range 5 {
 		require.NoError(t, tr.send([]byte("data"), "application/x-protobuf"))
 	}
-	assert.Equal(t, int64(1), atomic.LoadInt64(&connCount),
+	assert.Equal(t, int64(1), connCount.Load(),
 		"expected a single connection to be reused across sends")
 }