ddtrace/tracer: use map to keep track of spans started and finished

ddtrace/mocktracer/mockspan.go

@@ -105,7 +105,7 @@ type mockspan struct {
 	tags         map[string]interface{}
 	finishTime   time.Time
 	finished     bool
-	source       string
+	integration  string
 
 	startTime time.Time
 	parentID  uint64
@@ -133,7 +133,7 @@ func (s *mockspan) SetTag(key string, value interface{}) {
 		}
 	}
 	if key == ext.Component {
-		s.source = value.(string)
+		s.integration = value.(string)
 	}
 	s.tags[key] = value
 }
@@ -292,9 +292,7 @@ func (s *mockspan) Root() tracer.Span {
 	return root
 }
 
-// Source returns the component from which the mockspan was created.
-// This is used to test the source tag of the `datadog.tracer.spans_{started,finished}`
-// health metrics.
+// Integration returns the component from which the mockspan was created.
 func (s *mockspan) Integration() string {
-	return s.source
+	return s.integration
 }

ddtrace/mocktracer/mockspan_test.go

@@ -70,15 +70,15 @@ func TestNewSpan(t *testing.T) {
 		tr := new(mocktracer)
 		parentctx := &spanContext{spanID: 1, traceID: 2}
 		tags := make(map[string]interface{})
-		tags[ext.Component] = "sourceName"
+		tags[ext.Component] = "integrationName"
 		opts := &ddtrace.StartSpanConfig{Parent: parentctx, Tags: tags}
 		s := newSpan(tr, "opname", opts)
 
 		assert := assert.New(t)
 		assert.NotNil(s.context)
 		assert.Equal(uint64(1), s.parentID)
 		assert.Equal(uint64(2), s.context.traceID)
-		assert.Equal("sourceName", s.Integration())
+		assert.Equal("integrationName", s.Integration())
 	})
 }
 

ddtrace/tracer/metrics.go

@@ -6,6 +6,7 @@
 package tracer
 
 import (
+	"fmt"
 	"runtime"
 	"runtime/debug"
 	"sync/atomic"
@@ -91,8 +92,18 @@ func (t *tracer) reportHealthMetrics(interval time.Duration) {
 	for {
 		select {
 		case <-ticker.C:
-			t.statsd.Count("datadog.tracer.spans_started", int64(atomic.SwapUint32(&t.spansStarted, 0)), []string{"integration:manual"}, 1)
-			t.statsd.Count("datadog.tracer.spans_finished", int64(atomic.SwapUint32(&t.spansFinished, 0)), []string{"integration:manual"}, 1)
+			t.spansStarted.mu.Lock()
+			for name, v := range t.spansStarted.spans {
+				tag := fmt.Sprintf("integration:%s", name)
+				t.statsd.Count("datadog.tracer.spans_started", int64(v), []string{tag}, 1)
+			}
+			t.spansStarted.mu.Unlock()
+			t.spansFinished.mu.Lock()
+			for name, v := range t.spansFinished.spans {
+				tag := fmt.Sprintf("integration:%s", name)
+				t.statsd.Count("datadog.tracer.spans_finished", int64(v), []string{tag}, 1)
+			}
+			t.spansFinished.mu.Unlock()
 			t.statsd.Count("datadog.tracer.traces_dropped", int64(atomic.SwapUint32(&t.tracesDropped, 0)), []string{"reason:trace_too_large"}, 1)
 		case <-t.stop:
 			return

ddtrace/tracer/metrics_test.go

@@ -60,8 +60,8 @@ func TestReportHealthMetrics(t *testing.T) {
 	tg.Wait(assert, 3, 10*time.Second)
 
 	counts := tg.Counts()
-	assert.Equal(int64(1), counts["datadog.tracer.spans_started"])
-	assert.Equal(int64(1), counts["datadog.tracer.spans_finished"])
+	assert.GreaterOrEqual(counts["datadog.tracer.spans_started"], int64(1))
+	assert.GreaterOrEqual(counts["datadog.tracer.spans_finished"], int64(1))
 	assert.Equal(int64(0), counts["datadog.tracer.traces_dropped"])
 }
 
@@ -80,7 +80,7 @@ func TestSpansStartedTags(t *testing.T) {
 		tg.Wait(assert, 1, 100*time.Millisecond)
 
 		counts := tg.Counts()
-		assert.Equal(int64(1), counts["datadog.tracer.spans_started"])
+		assert.GreaterOrEqual(counts["datadog.tracer.spans_started"], int64(1))
 		for _, c := range tg.CountCalls() {
 			if c.GetName() != "datadog.tracer.spans_started" {
 				continue
@@ -104,7 +104,7 @@ func TestSpansStartedTags(t *testing.T) {
 		tg.Wait(assert, 1, 100*time.Millisecond)
 
 		counts := tg.Counts()
-		assert.Equal(int64(1), counts["datadog.tracer.spans_started"])
+		assert.GreaterOrEqual(counts["datadog.tracer.spans_started"], int64(1))
 		for _, c := range tg.CountCalls() {
 			if c.GetName() != "datadog.tracer.spans_started" {
 				continue
@@ -133,7 +133,7 @@ func TestSpansFinishedTags(t *testing.T) {
 		tg.Wait(assert, 1, 100*time.Millisecond)
 
 		counts := tg.Counts()
-		assert.Equal(int64(1), counts["datadog.tracer.spans_finished"])
+		assert.GreaterOrEqual(counts["datadog.tracer.spans_finished"], int64(1))
 		for _, c := range tg.CountCalls() {
 			if c.GetName() != "datadog.tracer.spans_finished" {
 				continue
@@ -156,7 +156,7 @@ func TestSpansFinishedTags(t *testing.T) {
 		tg.Wait(assert, 1, 100*time.Millisecond)
 
 		counts := tg.Counts()
-		assert.Equal(int64(1), counts["datadog.tracer.spans_finished"])
+		assert.GreaterOrEqual(counts["datadog.tracer.spans_finished"], int64(1))
 		for _, c := range tg.CountCalls() {
 			if c.GetName() != "datadog.tracer.spans_finished" {
 				continue

ddtrace/tracer/spancontext.go

@@ -531,11 +531,14 @@ func (t *trace) finishChunk(tr *tracer, ch *chunk) {
 		if sp == nil {
 			continue
 		}
-		if sp.integration == "manual" {
-			atomic.AddUint32(&tr.spansFinished, 1)
-		} else {
-			tr.statsd.Count("datadog.tracer.spans_finished", 1, []string{fmt.Sprintf("integration:%s", sp.integration)}, 1)
+		if tr.spansFinished.spans == nil {
+			tr.spansFinished.spans = make(map[string]uint32)
 		}
+		tr.spansFinished.mu.Lock()
+		count := tr.spansFinished.spans[sp.integration]
+		atomic.AddUint32(&count, 1)
+		tr.spansFinished.spans[sp.integration] = count
+		tr.spansFinished.mu.Unlock()
 	}
 	tr.pushChunk(ch)
 	t.finished = 0 // important, because a buffer can be used for several flushes

ddtrace/tracer/tracer.go

@@ -8,7 +8,6 @@ package tracer
 import (
 	gocontext "context"
 	"encoding/binary"
-	"fmt"
 	"log/slog"
 	"math"
 	"os"
@@ -80,9 +79,15 @@ type tracer struct {
 	// pid of the process
 	pid int
 
-	// These integers track metrics about spans and traces as they are started,
-	// finished, and dropped
-	spansStarted, spansFinished, tracesDropped uint32
+	// These maps keep count of the number of spans started and finished from
+	// each component, including contribs and "manual" spans.
+	spansStarted, spansFinished struct {
+		mu    sync.Mutex
+		spans map[string]uint32
+	}
+
+	// tracesDropped track metrics about traces as they are dropped
+	tracesDropped uint32
 
 	// Keeps track of the total number of traces dropped for accurate logging.
 	totalTracesDropped uint32
@@ -649,11 +654,14 @@ func (t *tracer) StartSpan(operationName string, options ...ddtrace.StartSpanOpt
 			log.Error("Abandoned spans channel full, disregarding span.")
 		}
 	}
-	if span.integration == "manual" {
-		atomic.AddUint32(&t.spansStarted, 1)
-	} else {
-		t.statsd.Count("datadog.tracer.spans_started", 1, []string{fmt.Sprintf("integration:%s", span.integration)}, 1)
+	if t.spansStarted.spans == nil {
+		t.spansStarted.spans = make(map[string]uint32)
 	}
+	t.spansStarted.mu.Lock()
+	defer t.spansStarted.mu.Unlock()
+	count := t.spansStarted.spans[span.integration]
+	atomic.AddUint32(&count, 1)
+	t.spansStarted.spans[span.integration] = count
 	return span
 }