ddtrace/tracer: initialize runtimeMetricsV2 with statsd "direct" client

DataDog · Dec 4, 2024 · f7fdf31 · f7fdf31
1 parent 076462a
commit f7fdf31
Show file tree

Hide file tree

Showing 8 changed files with 32 additions and 24 deletions.
diff --git a/contrib/database/sql/option_test.go b/contrib/database/sql/option_test.go
@@ -69,7 +69,7 @@ func TestCheckStatsdRequired(t *testing.T) {
 		cfg := new(config)
 		cfg.dbStats = true
 		cfg.checkStatsdRequired()
-		_, ok := cfg.statsdClient.(*statsd.Client)
+		_, ok := cfg.statsdClient.(*statsd.ClientDirect)
 		assert.True(t, ok)
 	})
 	t.Run("invalid address", func(t *testing.T) {

diff --git a/ddtrace/mocktracer/mocktracer.go b/ddtrace/mocktracer/mocktracer.go
@@ -80,7 +80,7 @@ func newMockTracer() *mocktracer {
 	client := &http.Client{
 		Transport: t.dsmTransport,
 	}
-	t.dsmProcessor = datastreams.NewProcessor(&statsd.NoOpClient{}, "env", "service", "v1", &url.URL{Scheme: "http", Host: "agent-address"}, client)
+	t.dsmProcessor = datastreams.NewProcessor(&statsd.NoOpClientDirect{}, "env", "service", "v1", &url.URL{Scheme: "http", Host: "agent-address"}, client)
 	t.dsmProcessor.Start()
 	t.dsmProcessor.Flush()
 	return &t

diff --git a/ddtrace/tracer/option.go b/ddtrace/tracer/option.go
@@ -799,7 +799,7 @@ func statsTags(c *config) []string {
 // withNoopStats is used for testing to disable statsd client
 func withNoopStats() StartOption {
 	return func(c *config) {
-		c.statsdClient = &statsd.NoOpClient{}
+		c.statsdClient = &statsd.NoOpClientDirect{}
 	}
 }
 

diff --git a/ddtrace/tracer/stats.go b/ddtrace/tracer/stats.go
@@ -131,7 +131,7 @@ func (c *concentrator) runFlusher(tick <-chan time.Time) {
 // statsd returns any tracer configured statsd client, or a no-op.
 func (c *concentrator) statsd() internal.StatsdClient {
 	if c.statsdClient == nil {
-		return &statsd.NoOpClient{}
+		return &statsd.NoOpClientDirect{}
 	}
 	return c.statsdClient
 }

diff --git a/ddtrace/tracer/textmap_test.go b/ddtrace/tracer/textmap_test.go
@@ -669,7 +669,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, test := range tests {
 				t.Run(fmt.Sprintf("inject with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					root := tracer.StartSpan("web.request").(*span)
 					ctx, ok := root.Context().(*spanContext)
@@ -741,7 +741,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, test := range tests {
 				t.Run(fmt.Sprintf("extract with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(test.in)
@@ -780,7 +780,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, tc := range tests {
 				t.Run(fmt.Sprintf("extract with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					_, err := tracer.Extract(tc.in)
@@ -837,7 +837,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, tc := range tests {
 				t.Run(fmt.Sprintf("extract with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(tc.in)
@@ -874,7 +874,7 @@ func TestEnvVars(t *testing.T) {
 		}
 		for i, tc := range tests {
 			t.Run(fmt.Sprintf("b3 single header inject #%d", i), func(t *testing.T) {
-				tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+				tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 				defer tracer.Stop()
 				root := tracer.StartSpan("myrequest").(*span)
 				ctx, ok := root.Context().(*spanContext)
@@ -932,7 +932,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, tc := range tests {
 				t.Run(fmt.Sprintf("inject with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithPropagator(NewPropagator(&PropagatorConfig{B3: true})), WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithPropagator(NewPropagator(&PropagatorConfig{B3: true})), WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					root := tracer.StartSpan("web.request").(*span)
 					ctx, ok := root.Context().(*spanContext)
@@ -1007,7 +1007,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, tc := range tests {
 				t.Run(fmt.Sprintf("extract with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 
@@ -1067,7 +1067,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for _, tc := range tests {
 				t.Run(fmt.Sprintf("inject and extract with env=%q", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					root := tracer.StartSpan("web.request").(*span)
 					root.SetTag(ext.SamplingPriority, -1)
@@ -1279,7 +1279,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%v extract/valid  with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(tc.in)
@@ -1335,7 +1335,7 @@ func TestEnvVars(t *testing.T) {
 
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%v extract/invalid  with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(tc)
@@ -1387,7 +1387,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%v extract/valid  with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(tc.inHeaders)
@@ -1589,7 +1589,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%d w3c inject with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					root := tracer.StartSpan("web.request").(*span)
@@ -1619,7 +1619,7 @@ func TestEnvVars(t *testing.T) {
 				})
 
 				t.Run(fmt.Sprintf("w3c inject with env=%q / testing tag list-member limit", testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					root := tracer.StartSpan("web.request").(*span)
@@ -1687,7 +1687,7 @@ func TestEnvVars(t *testing.T) {
 		}
 		for i, tc := range tests {
 			t.Run(fmt.Sprintf("#%d", i), func(t *testing.T) {
-				tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+				tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 				defer tracer.Stop()
 				assert := assert.New(t)
 				ctx, err := tracer.Extract(tc.inHeaders)
@@ -1762,7 +1762,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%d w3c inject/extract with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(tc.in)
@@ -1826,7 +1826,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%d w3c inject/extract with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					pCtx, err := tracer.Extract(tc.in)
@@ -1909,7 +1909,7 @@ func TestEnvVars(t *testing.T) {
 			}
 			for i, tc := range tests {
 				t.Run(fmt.Sprintf("#%v extract with env=%q", i, testEnv), func(t *testing.T) {
-					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClient{}))
+					tracer := newTracer(WithHTTPClient(c), withStatsdClient(&statsd.NoOpClientDirect{}))
 					defer tracer.Stop()
 					assert := assert.New(t)
 					ctx, err := tracer.Extract(tc.in)

diff --git a/internal/datastreams/processor_test.go b/internal/datastreams/processor_test.go
@@ -264,7 +264,7 @@ func BenchmarkSetCheckpoint(b *testing.B) {
 	client := &http.Client{
 		Transport: &noOpTransport{},
 	}
-	p := NewProcessor(&statsd.NoOpClient{}, "env", "service", "v1", &url.URL{Scheme: "http", Host: "agent-address"}, client)
+	p := NewProcessor(&statsd.NoOpClientDirect{}, "env", "service", "v1", &url.URL{Scheme: "http", Host: "agent-address"}, client)
 	p.Start()
 	for i := 0; i < b.N; i++ {
 		p.SetCheckpointWithParams(context.Background(), options.CheckpointParams{PayloadSize: 1000}, "type:edge-1", "direction:in", "type:kafka", "topic:topic1", "group:group1")

diff --git a/internal/statsd.go b/internal/statsd.go
@@ -19,6 +19,7 @@ type StatsdClient interface {
 	CountWithTimestamp(name string, value int64, tags []string, rate float64, timestamp time.Time) error
 	Gauge(name string, value float64, tags []string, rate float64) error
 	GaugeWithTimestamp(name string, value float64, tags []string, rate float64, timestamp time.Time) error
+	DistributionSamples(name string, values []float64, tags []string, rate float64) error
 	Timing(name string, value time.Duration, tags []string, rate float64) error
 	Flush() error
 	Close() error
@@ -29,9 +30,9 @@ func NewStatsdClient(addr string, globalTags []string) (StatsdClient, error) {
 	if addr == "" {
 		addr = DefaultDogstatsdAddr
 	}
-	client, err := statsd.New(addr, statsd.WithMaxMessagesPerPayload(40), statsd.WithTags(globalTags))
+	client, err := statsd.NewDirect(addr, statsd.WithMaxMessagesPerPayload(40), statsd.WithTags(globalTags))
 	if err != nil {
-		return &statsd.NoOpClient{}, err
+		return &statsd.NoOpClientDirect{}, err
 	}
 	return client, nil
 }
diff --git a/internal/statsdtest/statsdtest.go b/internal/statsdtest/statsdtest.go
@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"gopkg.in/DataDog/dd-trace-go.v1/internal"
 )
 
 type callType int64
@@ -24,6 +25,8 @@ const (
 	callTypeTiming
 )
 
+var _ internal.StatsdClient = &TestStatsdClient{}
+
 type TestStatsdClient struct {
 	mu          sync.RWMutex
 	gaugeCalls  []TestStatsdCall
@@ -104,6 +107,10 @@ func (tg *TestStatsdClient) CountWithTimestamp(name string, value int64, tags []
 	})
 }
 
+func (tg *TestStatsdClient) DistributionSamples(_ string, _ []float64, _ []string, _ float64) error {
+	panic("not implemented")
+}
+
 func (tg *TestStatsdClient) Timing(name string, value time.Duration, tags []string, rate float64) error {
 	return tg.addMetric(callTypeTiming, tags, TestStatsdCall{
 		name:    name,