Skip to content

Commit d6dc3b9

Browse files
mtoffl01kakkoyunmoezein0darccio
authored
fix(tracer): skip stats concentrator in OTLP export mode (#4609)
### What does this PR do? Introduces a `statsConcentrator` interface with a `noopConcentrator` implementation so the tracer skips stats computation in OTLP export mode. - `tracer.stats` is now a `statsConcentrator` interface instead of a concrete `*concentrator.` In OTLP mode it's assigned a `noopConcentrator`; otherwise a real concentrator. - The non-blocking channel send to `concentrator.In` was moved into a new `trySendSpan` method since interface fields can't expose struct fields — this also encapsulates the channel as an internal detail. ### Motivation: In OTLP export mode, Datadog-format stats shouldn't be computed or sent. This approach uses a no-op behind an interface so the disabled case is type-safe and can't be broken by forgetting a guard. ### Reviewer's Checklist <!-- * Authors can use this list as a reference to ensure that there are no problems during the review but the signing off is to be done by the reviewer(s). --> - [ ] Changed code has unit tests for its functionality at or near 100% coverage. - [ ] [System-Tests](https://github.com/DataDog/system-tests/) covering this feature have been added and enabled with the va.b.c-dev version tag. - [ ] There is a benchmark for any new code, or changes to existing code. - [ ] If this interacts with the agent in a new way, a system test has been added. - [ ] New code is free of linting errors. You can check this by running `make lint` locally. - [ ] New code doesn't break existing tests. You can check this by running `make test` locally. - [ ] Add an appropriate team label so this PR gets put in the right place for the release notes. - [ ] All generated files are up to date. You can check this by running `make generate` locally. - [ ] Non-trivial go.mod changes, e.g. adding new modules, are reviewed by @DataDog/dd-trace-go-guild. Make sure all nested modules are up to date by running `make fix-modules` locally. Unsure? Have a question? Request a review! Co-authored-by: kakkoyun <[email protected]> Co-authored-by: moezein0 <[email protected]> Co-authored-by: darccio <[email protected]>
1 parent 6114618 commit d6dc3b9

4 files changed

Lines changed: 132 additions & 10 deletions

File tree

‎ddtrace/tracer/stats.go‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,16 @@ var tracerObfuscationVersion = 1
3131
// covered in one stats bucket.
3232
var defaultStatsBucketSize = (10 * time.Second).Nanoseconds()
3333

34+
// statsConcentrator abstracts the stats-computation lifecycle so that callers
35+
// don't need nil checks when stats are disabled (e.g. OTLP export mode).
36+
type statsConcentrator interface {
37+
Start()
38+
Stop()
39+
flushAndSend(now time.Time, includeCurrent bool)
40+
newTracerStatSpan(s *Span, obfuscator *obfuscate.Obfuscator) (*tracerStatSpan, bool)
41+
trySendSpan(s *tracerStatSpan)
42+
}
43+
3444
// concentrator aggregates and stores statistics on incoming spans in time buckets,
3545
// flushing them occasionally to the underlying transport located in the given
3646
// tracer config.
@@ -269,3 +279,25 @@ func (c *concentrator) flushAndSend(timenow time.Time, includeCurrent bool) {
269279
}
270280
c.statsd().Incr("datadog.tracer.stats.flush_buckets", nil, float64(flushedBuckets))
271281
}
282+
283+
// trySendSpan attempts a non-blocking send of the stat span to the
284+
// concentrator's input channel.
285+
func (c *concentrator) trySendSpan(s *tracerStatSpan) {
286+
select {
287+
case c.In <- s:
288+
default:
289+
log.Error("Stats channel full, disregarding span.")
290+
}
291+
}
292+
293+
// noopConcentrator is a no-op implementation of statsConcentrator used when
294+
// client-side stats are disabled (e.g. OTLP export mode).
295+
type noopConcentrator struct{}
296+
297+
func (c *noopConcentrator) Start() {}
298+
func (c *noopConcentrator) Stop() {}
299+
func (c *noopConcentrator) flushAndSend(_ time.Time, _ bool) {}
300+
func (c *noopConcentrator) newTracerStatSpan(_ *Span, _ *obfuscate.Obfuscator) (*tracerStatSpan, bool) {
301+
return nil, false
302+
}
303+
func (c *noopConcentrator) trySendSpan(_ *tracerStatSpan) {}

‎ddtrace/tracer/stats_test.go‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -476,3 +476,37 @@ func TestStatsFlushRetries(t *testing.T) {
476476
})
477477
}
478478
}
479+
480+
func TestNoopConcentrator(t *testing.T) {
481+
var c statsConcentrator = &noopConcentrator{}
482+
483+
t.Run("Start", func(t *testing.T) {
484+
assert.NotPanics(t, func() { c.Start() })
485+
})
486+
487+
t.Run("Stop", func(t *testing.T) {
488+
assert.NotPanics(t, func() { c.Stop() })
489+
})
490+
491+
t.Run("flushAndSend", func(t *testing.T) {
492+
assert.NotPanics(t, func() { c.flushAndSend(time.Now(), false) })
493+
})
494+
495+
t.Run("newTracerStatSpan", func(t *testing.T) {
496+
s := &Span{
497+
name: "test.op",
498+
service: "test-service",
499+
resource: "/test",
500+
spanType: "web",
501+
start: time.Now().UnixNano(),
502+
duration: 1,
503+
}
504+
ss, ok := c.newTracerStatSpan(s, obfuscate.NewObfuscator(obfuscate.Config{}))
505+
assert.Nil(t, ss)
506+
assert.False(t, ok)
507+
})
508+
509+
t.Run("trySendSpan", func(t *testing.T) {
510+
assert.NotPanics(t, func() { c.trySendSpan(&tracerStatSpan{}) })
511+
})
512+
}

‎ddtrace/tracer/tracer.go‎

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -104,8 +104,8 @@ type tracer struct {
104104
config *config
105105

106106
// stats specifies the concentrator used to compute statistics, when client-side
107-
// stats are enabled.
108-
stats *concentrator
107+
// stats are enabled. In OTLP export mode this is a noopConcentrator.
108+
stats statsConcentrator
109109

110110
// traceWriter is responsible for sending finished traces to their
111111
// destination, such as the Trace Agent or Datadog Forwarder.
@@ -497,6 +497,10 @@ func newUnstartedTracer(opts ...StartOption) (t *tracer, err error) {
497497
c.internalConfig.SetLogDirectory("", telemetry.OriginCalculated)
498498
}
499499
}
500+
var sc statsConcentrator = newConcentrator(c, defaultStatsBucketSize, statsd)
501+
if c.internalConfig.OTLPExportMode() {
502+
sc = &noopConcentrator{}
503+
}
500504
t = &tracer{
501505
config: c,
502506
traceWriter: writer,
@@ -507,7 +511,7 @@ func newUnstartedTracer(opts ...StartOption) (t *tracer, err error) {
507511
defaultSampler: dfltSampler,
508512
pid: os.Getpid(),
509513
logDroppedTraces: time.NewTicker(1 * time.Second),
510-
stats: newConcentrator(c, defaultStatsBucketSize, statsd),
514+
stats: sc,
511515
spansStarted: *globalinternal.NewXSyncMapCounterMap(),
512516
spansFinished: *globalinternal.NewXSyncMapCounterMap(),
513517
obfuscator: obfuscate.NewObfuscator(func() obfuscate.Config {
@@ -1163,13 +1167,7 @@ func (t *tracer) submit(s *Span) {
11631167
if !shouldCalc {
11641168
return
11651169
}
1166-
// the agent supports computed stats
1167-
select {
1168-
case t.stats.In <- statSpan:
1169-
// ok
1170-
default:
1171-
log.Error("Stats channel full, disregarding span.")
1172-
}
1170+
t.stats.trySendSpan(statSpan)
11731171
}
11741172

11751173
func (t *tracer) submitAbandonedSpan(s *Span, finished bool) {

‎ddtrace/tracer/tracer_test.go‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,9 @@ import (
2828
"testing/synctest"
2929
"time"
3030

31+
otlptrace "go.opentelemetry.io/proto/otlp/trace/v1"
3132
"go.uber.org/goleak"
33+
"google.golang.org/protobuf/proto"
3234

3335
"github.com/DataDog/dd-trace-go/v2/ddtrace"
3436
"github.com/DataDog/dd-trace-go/v2/ddtrace/ext"
@@ -1495,6 +1497,7 @@ func TestOTLPExportMode(t *testing.T) {
14951497
assert.True(isOTLPWriter, "expected otlpTraceWriter in OTLP export mode")
14961498
_, isAlwaysOn := tracer.defaultSampler.(*otelParentBasedAlwaysOnSampler)
14971499
assert.True(isAlwaysOn, "expected otelParentBasedAlwaysOnSampler in OTLP export mode")
1500+
assert.IsType(&noopConcentrator{}, tracer.stats, "expected noopConcentrator in OTLP export mode")
14981501
})
14991502

15001503
t.Run("OTEL_TRACES_EXPORTER=otlp env var enables OTLP mode", func(t *testing.T) {
@@ -1510,6 +1513,61 @@ func TestOTLPExportMode(t *testing.T) {
15101513
})
15111514
}
15121515

1516+
func TestOTLPExportModeStatsSkipped(t *testing.T) {
1517+
srv := newTestOTLPServer()
1518+
defer srv.Close()
1519+
1520+
tick := make(chan time.Time)
1521+
trc, err := newTracer(
1522+
func(c *config) {
1523+
c.internalConfig.SetOTLPExportMode(true, internalconfig.OriginCode)
1524+
c.ddTransport = newDummyTransport()
1525+
c.tickChan = tick
1526+
},
1527+
)
1528+
require.NoError(t, err)
1529+
1530+
w := trc.traceWriter.(*otlpTraceWriter)
1531+
w.transport = newOTLPTransport(srv.Client(), srv.URL, map[string]string{"Content-Type": "application/x-protobuf"})
1532+
1533+
// Enable canDropP0s so submit() reaches the noopConcentrator
1534+
// rather than short-circuiting.
1535+
af := trc.config.agent.load()
1536+
af.Stats = true
1537+
af.DropP0s = true
1538+
trc.config.agent.store(af)
1539+
trc.config.internalConfig.SetFeatureFlags([]string{"discovery"}, internalconfig.OriginCode)
1540+
1541+
setGlobalTracer(trc)
1542+
1543+
assert.IsType(t, &noopConcentrator{}, trc.stats, "concentrator must be noop in OTLP mode")
1544+
assert.True(t, trc.config.canDropP0s(), "canDropP0s must be true for this test to exercise submit()")
1545+
1546+
const spanCount = 5
1547+
for range spanCount {
1548+
span := trc.newRootSpan("test.op", "test-service", "/test")
1549+
span.Finish()
1550+
}
1551+
1552+
// Stop drains t.out, flushes the writer, and waits for in-flight sends.
1553+
trc.Stop()
1554+
1555+
payloads := srv.getPayloads()
1556+
require.NotEmpty(t, payloads, "expected at least one OTLP payload")
1557+
1558+
totalSpans := 0
1559+
for _, p := range payloads {
1560+
var td otlptrace.TracesData
1561+
require.NoError(t, proto.Unmarshal(p, &td))
1562+
for _, rs := range td.ResourceSpans {
1563+
for _, ss := range rs.ScopeSpans {
1564+
totalSpans += len(ss.Spans)
1565+
}
1566+
}
1567+
}
1568+
assert.Equal(t, spanCount, totalSpans, "all spans should be retained in OTLP mode, not dropped by nil concentrator")
1569+
}
1570+
15131571
func TestTracerConcurrent(t *testing.T) {
15141572
assert := assert.New(t)
15151573
tracer, transport, flush, stop, err := startTestTracer(t)

0 commit comments

Comments
 (0)