stats need accurate origin for synthetics

ddtrace/tracer/stats.go

@@ -31,7 +31,7 @@ type concentrator struct {
 	// In specifies the channel to be used for feeding data to the concentrator.
 	// In order for In to have a consumer, the concentrator must be started using
 	// a call to Start.
-	In chan *stats.StatSpan
+	In chan *tracerStatSpan
 
 	// stopped reports whether the concentrator is stopped (when non-zero)
 	stopped uint32
@@ -47,6 +47,11 @@ type concentrator struct {
 	statsdClient internal.StatsdClient // statsd client for sending metrics.
 }
 
+type tracerStatSpan struct {
+	statSpan *stats.StatSpan
+	origin   string
+}
+
 // newConcentrator creates a new concentrator using the given tracer
 // configuration c. It creates buckets of bucketSize nanoseconds duration.
 func newConcentrator(c *config, bucketSize int64) *concentrator {
@@ -75,7 +80,7 @@ func newConcentrator(c *config, bucketSize int64) *concentrator {
 	}
 	spanConcentrator := stats.NewSpanConcentrator(sCfg, time.Now())
 	return &concentrator{
-		In:               make(chan *stats.StatSpan, 10000),
+		In:               make(chan *tracerStatSpan, 10000),
 		bucketSize:       bucketSize,
 		stopped:          1,
 		cfg:              c,
@@ -145,14 +150,22 @@ func (c *concentrator) runIngester() {
 	}
 }
 
-func (c *concentrator) newAggregableSpan(s *span, obfuscator *obfuscate.Obfuscator) (*stats.StatSpan, bool) {
-	return c.spanConcentrator.NewStatSpan(s.Service, obfuscatedResource(obfuscator, s.Type, s.Resource),
+func (c *concentrator) newAggregableSpan(s *span, obfuscator *obfuscate.Obfuscator) (*tracerStatSpan, bool) {
+	statSpan, ok := c.spanConcentrator.NewStatSpan(s.Service, obfuscatedResource(obfuscator, s.Type, s.Resource),
 		s.Name, s.Type, s.ParentID, s.Start, s.Duration, s.Error, s.Meta, s.Metrics, c.cfg.agent.peerTags)
+	if !ok {
+		return nil, false
+	}
+	origin := s.Meta[keyOrigin]
+	return &tracerStatSpan{
+		statSpan: statSpan,
+		origin:   origin,
+	}, true
 }
 
 // add adds s into the concentrator's internal stats buckets.
-func (c *concentrator) add(s *stats.StatSpan) {
-	c.spanConcentrator.AddSpan(s, c.aggregationKey, "", nil, "") //todo: origin?
+func (c *concentrator) add(s *tracerStatSpan) {
+	c.spanConcentrator.AddSpan(s.statSpan, c.aggregationKey, "", nil, s.origin)
 }
 
 // Stop stops the concentrator and blocks until the operation completes.

ddtrace/tracer/stats_test.go

@@ -11,9 +11,6 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-
-	"github.com/DataDog/datadog-agent/pkg/trace/stats"
 )
 
 func TestAlignTs(t *testing.T) {
@@ -25,22 +22,21 @@ func TestAlignTs(t *testing.T) {
 
 func TestConcentrator(t *testing.T) {
 	bucketSize := int64(500_000)
-
-	sc := stats.NewSpanConcentrator(&stats.SpanConcentratorConfig{BucketInterval: bucketSize}, time.Now())
-	ss1, ok := sc.NewStatSpan("", "", "http.request", "", 0,
-		time.Now().UnixNano()+3*bucketSize,
-		1,
-		0, nil, map[string]float64{keyMeasured: 1}, nil)
-	require.True(t, ok)
-	ss2, ok := sc.NewStatSpan("", "", "sql.query", "", 0,
-		time.Now().UnixNano()+4*bucketSize,
-		1,
-		0, nil, map[string]float64{keyMeasured: 1}, nil)
-	require.True(t, ok)
-
+	s1 := span{
+		Name:     "http.request",
+		Start:    time.Now().UnixNano() + 3*bucketSize,
+		Duration: 1,
+		Metrics:  map[string]float64{keyMeasured: 1},
+	}
+	s2 := span{
+		Name:     "sql.query",
+		Start:    time.Now().UnixNano() + 4*bucketSize,
+		Duration: 1,
+		Metrics:  map[string]float64{keyMeasured: 1},
+	}
 	t.Run("start-stop", func(t *testing.T) {
 		assert := assert.New(t)
-		c := newConcentrator(&config{}, defaultStatsBucketSize)
+		c := newConcentrator(&config{}, bucketSize)
 		assert.EqualValues(atomic.LoadUint32(&c.stopped), 1)
 		c.Start()
 		assert.EqualValues(atomic.LoadUint32(&c.stopped), 0)
@@ -99,6 +95,8 @@ func TestConcentrator(t *testing.T) {
 			transport := newDummyTransport()
 			c := newConcentrator(&config{transport: transport, env: "someEnv"}, 500_000)
 			assert.Len(t, transport.Stats(), 0)
+			ss1, ok := c.newAggregableSpan(&s1, nil)
+			assert.True(t, ok)
 			c.Start()
 			c.In <- ss1
 			time.Sleep(2 * time.Millisecond * timeMultiplicator)
@@ -114,6 +112,10 @@ func TestConcentrator(t *testing.T) {
 			transport := newDummyTransport()
 			c := newConcentrator(&config{transport: transport, env: "someEnv"}, (10 * time.Second).Nanoseconds())
 			assert.Len(t, transport.Stats(), 0)
+			ss1, ok := c.newAggregableSpan(&s1, nil)
+			assert.True(t, ok)
+			ss2, ok := c.newAggregableSpan(&s2, nil)
+			assert.True(t, ok)
 			c.Start()
 			c.In <- ss1
 			c.In <- ss2
@@ -130,21 +132,18 @@ func TestConcentrator(t *testing.T) {
 			assert.NotNil(t, names["http.request"])
 			assert.NotNil(t, names["potato"])
 		})
+
+		// stats should be sent if the concentrator is stopped
+		t.Run("stop", func(t *testing.T) {
+			transport := newDummyTransport()
+			c := newConcentrator(&config{transport: transport}, 500000)
+			assert.Len(t, transport.Stats(), 0)
+			ss1, ok := c.newAggregableSpan(&s1, nil)
+			assert.True(t, ok)
+			c.Start()
+			c.In <- ss1
+			c.Stop()
+			assert.NotEmpty(t, transport.Stats())
+		})
 	})
-	//
-	//	// stats should be sent if the concentrator is stopped
-	//	t.Run("stop", func(t *testing.T) {
-	//		transport := newDummyTransport()
-	//		c := newConcentrator(&config{transport: transport}, 500000)
-	//		assert.Len(t, transport.Stats(), 0)
-	//		c.Start()
-	//		c.In <- &aggregableSpan{
-	//			key:      key1,
-	//			Start:    time.Now().UnixNano(),
-	//			Duration: 1,
-	//		}
-	//		c.Stop()
-	//		assert.NotEmpty(t, transport.Stats())
-	//	})
-	//})
 }