ddtrace/tracer: use xsync Map for spansStarted and Finished

ddtrace/tracer/metrics.go

@@ -93,18 +93,22 @@ func (t *tracer) reportHealthMetricsAtInterval(interval time.Duration) {
 	for {
 		select {
 		case <-ticker.C:
-			t.spansStarted.mu.Lock()
-			for name, v := range t.spansStarted.spans {
-				t.statsd.Count("datadog.tracer.spans_started", v, []string{"integration:" + name}, 1)
-				t.spansStarted.spans[name] = 0
-			}
-			t.spansStarted.mu.Unlock()
-			t.spansFinished.mu.Lock()
-			for name, v := range t.spansFinished.spans {
-				t.statsd.Count("datadog.tracer.spans_finished", v, []string{"integration:" + name}, 1)
-				t.spansFinished.spans[name] = 0
-			}
-			t.spansFinished.mu.Unlock()
+			// if there are started spans, report the number spans with their integration, then
+			// reset the count
+			t.spansStarted.Range(func(key string, value int64) bool {
+				err := t.statsd.Count("datadog.tracer.spans_started", value, []string{"integration:" + key}, 1)
+				return err == nil
+			})
+			t.spansStarted.Clear()
+
+			// if there are finished spans, report the number spans with their integration, then
+			// reset the count
+			t.spansFinished.Range(func(key string, value int64) bool {
+				err := t.statsd.Count("datadog.tracer.spans_finished", value, []string{"integration:" + key}, 1)
+				return err == nil
+			})
+			t.spansFinished.Clear()
+
 			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

@@ -13,6 +13,7 @@ import (
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
 	globalinternal "gopkg.in/DataDog/dd-trace-go.v1/internal"
 
+	"github.com/puzpuzpuz/xsync/v3"
 	"github.com/stretchr/testify/assert"
 
 	"gopkg.in/DataDog/dd-trace-go.v1/internal/statsdtest"
@@ -102,7 +103,7 @@ func TestSpansStartedTags(t *testing.T) {
 
 		tracer.StartSpan("operation").Finish()
 		tg.Wait(assert, 1, 100*time.Millisecond)
-		assertSpanMetricCountsAreZero(t, tracer.spansStarted.spans)
+		assertSpanMetricCountsAreZero(t, tracer.spansStarted)
 
 		counts := tg.Counts()
 		assert.Equal(counts["datadog.tracer.spans_started"], int64(1))
@@ -124,7 +125,7 @@ func TestSpansStartedTags(t *testing.T) {
 		defer sp.Finish()
 
 		tg.Wait(assert, 1, 100*time.Millisecond)
-		assertSpanMetricCountsAreZero(t, tracer.spansStarted.spans)
+		assertSpanMetricCountsAreZero(t, tracer.spansStarted)
 
 		counts := tg.Counts()
 		assert.Equal(counts["datadog.tracer.spans_started"], int64(1))
@@ -151,7 +152,7 @@ func TestSpansFinishedTags(t *testing.T) {
 
 		tracer.StartSpan("operation").Finish()
 		tg.Wait(assert, 1, 100*time.Millisecond)
-		assertSpanMetricCountsAreZero(t, tracer.spansFinished.spans)
+		assertSpanMetricCountsAreZero(t, tracer.spansFinished)
 
 		counts := tg.Counts()
 		assert.Equal(counts["datadog.tracer.spans_finished"], int64(1))
@@ -172,7 +173,7 @@ func TestSpansFinishedTags(t *testing.T) {
 		tracer.StartSpan("operation", Tag(ext.Component, "contrib")).Finish()
 
 		tg.Wait(assert, 1, 100*time.Millisecond)
-		assertSpanMetricCountsAreZero(t, tracer.spansFinished.spans)
+		assertSpanMetricCountsAreZero(t, tracer.spansFinished)
 
 		counts := tg.Counts()
 		assert.Equal(counts["datadog.tracer.spans_finished"], int64(1))
@@ -210,8 +211,8 @@ func TestHealthMetricsRaceCondition(t *testing.T) {
 	assert.Equal(int64(5), counts["datadog.tracer.spans_started"])
 	assert.Equal(int64(5), counts["datadog.tracer.spans_finished"])
 
-	assertSpanMetricCountsAreZero(t, tracer.spansStarted.spans)
-	assertSpanMetricCountsAreZero(t, tracer.spansFinished.spans)
+	assertSpanMetricCountsAreZero(t, tracer.spansStarted)
+	assertSpanMetricCountsAreZero(t, tracer.spansFinished)
 }
 
 func TestTracerMetrics(t *testing.T) {
@@ -254,8 +255,8 @@ func BenchmarkSpansMetrics(b *testing.B) {
 	}
 }
 
-func assertSpanMetricCountsAreZero(t *testing.T, metric map[string]int64) {
-	for _, v := range metric {
-		assert.Equal(t, int64(0), v)
-	}
+func assertSpanMetricCountsAreZero(t *testing.T, metric *xsync.MapOf[string, int64]) {
+	metric.Range(func(_ string, value int64) bool {
+		return assert.Equal(t, int64(0), value)
+	})
 }

ddtrace/tracer/spancontext.go

@@ -531,12 +531,11 @@ func (t *trace) finishChunk(tr *tracer, ch *chunk) {
 		if sp == nil {
 			continue
 		}
-		tr.spansFinished.mu.Lock()
-		if tr.spansFinished.spans == nil {
-			tr.spansFinished.spans = make(map[string]int64)
-		}
-		tr.spansFinished.spans[sp.integration] += 1
-		tr.spansFinished.mu.Unlock()
+		tr.spansFinished.Compute(sp.integration, func(oldValue int64, _ bool) (newValue int64, delete bool) {
+			newValue = oldValue + 1
+			delete = false
+			return
+		})
 	}
 	tr.pushChunk(ch)
 	t.finished = 0 // important, because a buffer can be used for several flushes

ddtrace/tracer/tracer.go

@@ -18,6 +18,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/puzpuzpuz/xsync/v3"
+
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/ext"
 	"gopkg.in/DataDog/dd-trace-go.v1/ddtrace/internal"
@@ -81,10 +83,7 @@ type tracer struct {
 
 	// 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]int64
-	}
+	spansStarted, spansFinished *xsync.MapOf[string, int64]
 
 	// tracesDropped track metrics about traces as they are dropped
 	tracesDropped uint32
@@ -323,6 +322,8 @@ func newUnstartedTracer(opts ...StartOption) *tracer {
 		pid:              os.Getpid(),
 		logDroppedTraces: time.NewTicker(1 * time.Second),
 		stats:            newConcentrator(c, defaultStatsBucketSize, statsd),
+		spansStarted:     xsync.NewMapOf[string, int64](),
+		spansFinished:    xsync.NewMapOf[string, int64](),
 		obfuscator: obfuscate.NewObfuscator(obfuscate.Config{
 			SQL: obfuscate.SQLConfig{
 				TableNames:       c.agent.HasFlag("table_names"),
@@ -671,12 +672,11 @@ func (t *tracer) StartSpan(operationName string, options ...ddtrace.StartSpanOpt
 			log.Error("Abandoned spans channel full, disregarding span.")
 		}
 	}
-	t.spansStarted.mu.Lock()
-	defer t.spansStarted.mu.Unlock()
-	if t.spansStarted.spans == nil {
-		t.spansStarted.spans = make(map[string]int64)
-	}
-	t.spansStarted.spans[span.integration] += 1
+	t.spansStarted.Compute(span.integration, func(oldValue int64, _ bool) (newValue int64, delete bool) {
+		newValue = oldValue + 1
+		delete = false
+		return
+	})
 	return span
 }
 

go.mod

@@ -246,6 +246,7 @@ require (
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c // indirect
+	github.com/puzpuzpuz/xsync/v3 v3.4.0
 	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
 	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
 	github.com/rivo/uniseg v0.4.4 // indirect

go.sum

@@ -1942,6 +1942,8 @@ github.com/prometheus/procfs v0.7.3/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1
 github.com/prometheus/procfs v0.15.0 h1:A82kmvXJq2jTu5YUhSGNlYoxh85zLnKgPz4bMZgI5Ek=
 github.com/prometheus/procfs v0.15.0/go.mod h1:Y0RJ/Y5g5wJpkTisOtqwDSo4HwhGmLB4VQSw2sQJLHk=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
+github.com/puzpuzpuz/xsync/v3 v3.4.0 h1:DuVBAdXuGFHv8adVXjWWZ63pJq+NRXOWVXlKDBZ+mJ4=
+github.com/puzpuzpuz/xsync/v3 v3.4.0/go.mod h1:VjzYrABPabuM4KyBh1Ftq6u8nhwY5tBPKP9jpmh0nnA=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 h1:N/ElC8H3+5XpJzTSTfLsJV/mx9Q9g7kxmchpfZyxgzM=
 github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
 github.com/redis/go-redis/v9 v9.1.0 h1:137FnGdk+EQdCbye1FW+qOEcY5S+SpY9T0NiuqvtfMY=