fix(tracer): skip stats concentrator in OTLP export mode

ddtrace/tracer/stats.go

@@ -31,6 +31,16 @@ var tracerObfuscationVersion = 1
 // covered in one stats bucket.
 var defaultStatsBucketSize = (10 * time.Second).Nanoseconds()
 
+// statsConcentrator abstracts the stats-computation lifecycle so that callers
+// don't need nil checks when stats are disabled (e.g. OTLP export mode).
+type statsConcentrator interface {
+	Start()
+	Stop()
+	flushAndSend(now time.Time, includeCurrent bool)
+	newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscator) (*tracerStatSpan, bool)
+	trySendSpan(s *tracerStatSpan)
+}
+
 // concentrator aggregates and stores statistics on incoming spans in time buckets,
 // flushing them occasionally to the underlying transport located in the given
 // tracer config.
@@ -269,3 +279,25 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) {
 	}
 	c.statsd().Incr("datadog.tracer.stats.flush_buckets", nil, float64(flushedBuckets))
 }
+
+// trySendSpan attempts a non-blocking send of the stat span to the
+// concentrator's input channel.
+func (c *concentrator) trySendSpan(s *tracerStatSpan) {
+	select {
+	case c.In <- s:
+	default:
+		log.Error("Stats channel full, disregarding span.")
+	}
+}
+
+// noopConcentrator is a no-op implementation of statsConcentrator used when
+// client-side stats are disabled (e.g. OTLP export mode).
+type noopConcentrator struct{}
+
+func (c *noopConcentrator) Start()                           {}
+func (c *noopConcentrator) Stop()                            {}
+func (c *noopConcentrator) flushAndSend(_ time.Time, _ bool) {}
+func (c *noopConcentrator) newTracerStatSpan(_ *Span, _ *obfuscate.Obfuscator) (*tracerStatSpan, bool) {
+	return nil, false
+}
+func (c *noopConcentrator) trySendSpan(_ *tracerStatSpan) {}

ddtrace/tracer/stats_test.go

@@ -476,3 +476,37 @@ func TestStatsFlushRetries(t *testing.T) {
 		})
 	}
 }
+
+func TestNoopConcentrator(t *testing.T) {
+	var c statsConcentrator = &noopConcentrator{}
+
+	t.Run("Start", func(t *testing.T) {
+		assert.NotPanics(t, func() { c.Start() })
+	})
+
+	t.Run("Stop", func(t *testing.T) {
+		assert.NotPanics(t, func() { c.Stop() })
+	})
+
+	t.Run("flushAndSend", func(t *testing.T) {
+		assert.NotPanics(t, func() { c.flushAndSend(time.Now(), false) })
+	})
+
+	t.Run("newTracerStatSpan", func(t *testing.T) {
+		s := &Span{
+			name:     "test.op",
+			service:  "test-service",
+			resource: "/test",
+			spanType: "web",
+			start:    time.Now().UnixNano(),
+			duration: 1,
+		}
+		ss, ok := c.newTracerStatSpan(s, obfuscate.NewObfuscator(obfuscate.Config{}))
+		assert.Nil(t, ss)
+		assert.False(t, ok)
+	})
+
+	t.Run("trySendSpan", func(t *testing.T) {
+		assert.NotPanics(t, func() { c.trySendSpan(&tracerStatSpan{}) })
+	})
+}

ddtrace/tracer/tracer.go

@@ -104,8 +104,8 @@ type tracer struct {
 	config *config
 
 	// stats specifies the concentrator used to compute statistics, when client-side
-	// stats are enabled.
-	stats *concentrator
+	// stats are enabled. In OTLP export mode this is a noopConcentrator.
+	stats statsConcentrator
 
 	// traceWriter is responsible for sending finished traces to their
 	// destination, such as the Trace Agent or Datadog Forwarder.
@@ -497,6 +497,10 @@ func newUnstartedTracer(opts ...StartOption) (t *tracer, err error) {
 			c.internalConfig.SetLogDirectory("", telemetry.OriginCalculated)
 		}
 	}
+	var sc statsConcentrator = newConcentrator(c, defaultStatsBucketSize, statsd)
+	if c.internalConfig.OTLPExportMode() {
+		sc = &noopConcentrator{}
+	}
 	t = &tracer{
 		config:           c,
 		traceWriter:      writer,
@@ -507,7 +511,7 @@ func newUnstartedTracer(opts ...StartOption) (t *tracer, err error) {
 		defaultSampler:   dfltSampler,
 		pid:              os.Getpid(),
 		logDroppedTraces: time.NewTicker(1 * time.Second),
-		stats:            newConcentrator(c, defaultStatsBucketSize, statsd),
+		stats:            sc,
 		spansStarted:     *globalinternal.NewXSyncMapCounterMap(),
 		spansFinished:    *globalinternal.NewXSyncMapCounterMap(),
 		obfuscator: obfuscate.NewObfuscator(func() obfuscate.Config {
@@ -1163,13 +1167,7 @@ func (t *tracer) submit(s *Span) {
 	if !shouldCalc {
 		return
 	}
-	// the agent supports computed stats
-	select {
-	case t.stats.In <- statSpan:
-		// ok
-	default:
-		log.Error("Stats channel full, disregarding span.")
-	}
+	t.stats.trySendSpan(statSpan)
 }
 
 func (t *tracer) submitAbandonedSpan(s *Span, finished bool) {

ddtrace/tracer/tracer_test.go

@@ -28,7 +28,9 @@ import (
 	"testing/synctest"
 	"time"
 
+	otlptrace "go.opentelemetry.io/proto/otlp/trace/v1"
 	"go.uber.org/goleak"
+	"google.golang.org/protobuf/proto"
 
 	"github.com/DataDog/dd-trace-go/v2/ddtrace"
 	"github.com/DataDog/dd-trace-go/v2/ddtrace/ext"
@@ -1495,6 +1497,7 @@ func TestOTLPExportMode(t *testing.T) {
 		assert.True(isOTLPWriter, "expected otlpTraceWriter in OTLP export mode")
 		_, isAlwaysOn := tracer.defaultSampler.(*otelParentBasedAlwaysOnSampler)
 		assert.True(isAlwaysOn, "expected otelParentBasedAlwaysOnSampler in OTLP export mode")
+		assert.IsType(&noopConcentrator{}, tracer.stats, "expected noopConcentrator in OTLP export mode")
 	})
 
 	t.Run("OTEL_TRACES_EXPORTER=otlp env var enables OTLP mode", func(t *testing.T) {
@@ -1510,6 +1513,61 @@ func TestOTLPExportMode(t *testing.T) {
 	})
 }
 
+func TestOTLPExportModeStatsSkipped(t *testing.T) {
+	srv := newTestOTLPServer()
+	defer srv.Close()
+
+	tick := make(chan time.Time)
+	trc, err := newTracer(
+		func(c *config) {
+			c.internalConfig.SetOTLPExportMode(true, internalconfig.OriginCode)
+			c.ddTransport = newDummyTransport()
+			c.tickChan = tick
+		},
+	)
+	require.NoError(t, err)
+
+	w := trc.traceWriter.(*otlpTraceWriter)
+	w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"})
+
+	// Enable canDropP0s so submit() reaches the noopConcentrator
+	// rather than short-circuiting.
+	af := trc.config.agent.load()
+	af.Stats = true
+	af.DropP0s = true
+	trc.config.agent.store(af)
+	trc.config.internalConfig.SetFeatureFlags([]string{"discovery"}, internalconfig.OriginCode)
+
+	setGlobalTracer(trc)
+
+	assert.IsType(t, &noopConcentrator{}, trc.stats, "concentrator must be noop in OTLP mode")
+	assert.True(t, trc.config.canDropP0s(), "canDropP0s must be true for this test to exercise submit()")
+
+	const spanCount = 5
+	for range spanCount {
+		span := trc.newRootSpan("test.op", "test-service", "/test")
+		span.Finish()
+	}
+
+	// Stop drains t.out, flushes the writer, and waits for in-flight sends.
+	trc.Stop()
+
+	payloads := srv.getPayloads()
+	require.NotEmpty(t, payloads, "expected at least one OTLP payload")
+
+	totalSpans := 0
+	for _, p := range payloads {
+		var td otlptrace.TracesData
+		require.NoError(t, proto.Unmarshal(p, &td))
+		for _, rs := range td.ResourceSpans {
+			for _, ss := range rs.ScopeSpans {
+				totalSpans += len(ss.Spans)
+			}
+		}
+	}
+	assert.Equal(t, spanCount, totalSpans, "all spans should be retained in OTLP mode, not dropped by nil concentrator")
+}
+
 func TestTracerConcurrent(t *testing.T) {
 	assert := assert.New(t)
 	tracer, transport, flush, stop, err := startTestTracer(t)