diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go index bb1fa8fbbaf..b230adb936d 100644 --- a/ddtrace/tracer/stats.go +++ b/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) {} diff --git a/ddtrace/tracer/stats_test.go b/ddtrace/tracer/stats_test.go index c3b772c82fd..59e4d357f35 100644 --- a/ddtrace/tracer/stats_test.go +++ b/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{}) }) + }) +} diff --git a/ddtrace/tracer/tracer.go b/ddtrace/tracer/tracer.go index 32c082503e0..076f1fda37a 100644 --- a/ddtrace/tracer/tracer.go +++ b/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) { diff --git a/ddtrace/tracer/tracer_test.go b/ddtrace/tracer/tracer_test.go index 3ff52e836cb..225a9f5d4bf 100644 --- a/ddtrace/tracer/tracer_test.go +++ b/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)