fix: data race on sender.enabled in trace writer (#54640)

<!--Please give us some feedback on your experience writing this PR ! https://app.datadoghq.com/forms/43db4c02-6837-400c-8083-692e141b1b88 !-->

### What does this PR do?

Fixes a data race on `sender.enabled` in `pkg/trace/writer/sender.go`: `isEnabled()` read/wrote the field with no synchronization. Switched it to `atomic.Bool`.

### Motivation

Fix a race, found via a race-detector-enabled build in staging.

### Describe how you validated your changes

Added `TestIsEnabledConcurrent` in `pkg/trace/writer/sender_test.go`, which calls `isEnabled()` concurrently; it reproduces the race under `-race` before the fix and is clean after.

### Additional Notes


Co-authored-by: pierre.gimalac <pierre.gimalac@datadoghq.com>

Author: pgimalac

pkg/trace/writer/BUILD.bazel

@@ -57,5 +57,6 @@ dd_agent_go_test(
         "@com_github_stretchr_testify//require",
         "@com_github_tinylib_msgp//msgp",
         "@org_golang_google_protobuf//proto",
+        "@org_uber_go_atomic//:atomic",
     ],
 )

pkg/trace/writer/sender.go

@@ -244,7 +244,7 @@ type sender struct {
 	mu      sync.RWMutex // guards closed
 	closed  bool         // closed reports if the loop is stopped
 	statsd  statsd.ClientInterface
-	enabled bool // false on inactive MRF senders. True otherwise
+	enabled *atomic.Bool // false on inactive MRF senders. True otherwise
 }
 
 // newSender returns a new sender based on the given config cfg.
@@ -256,7 +256,7 @@ func newSender(cfg *senderConfig, apiKeyManager *apiKeyManager, statsd statsd.Cl
 		inflight:      atomic.NewInt32(0),
 		maxRetries:    int32(cfg.maxRetries),
 		statsd:        statsd,
-		enabled:       true,
+		enabled:       atomic.NewBool(true),
 	}
 	for i := 0; i < cfg.maxConns; i++ {
 		go s.loop()
@@ -432,15 +432,13 @@ func (s *sender) isEnabled() bool {
 	}
 	// Endpoint is MRF and MRF is enabled. Figure out if we need to failover APM data
 	if s.cfg.MRFFailoverAPM() {
-		if !s.enabled {
+		if s.enabled.CompareAndSwap(false, true) {
 			log.Infof("Sender for domain %v has been failed over to, enabling it for MRF.", s.cfg.url)
-			s.enabled = true
 		}
 		return true
 	}
 
-	if s.enabled {
-		s.enabled = false
+	if s.enabled.CompareAndSwap(true, false) {
 		log.Infof("Sender for domain %v was disabled; payloads will be dropped for this domain.", s.cfg.url)
 	} else {
 		log.Debugf("Sender for domain %v is disabled; dropping payload for this domain.", s.cfg.url)

pkg/trace/writer/sender_test.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
+	"go.uber.org/atomic"
 
 	"github.com/DataDog/datadog-go/v5/statsd"
 
@@ -457,6 +458,31 @@ func syncTest403ThrottlesRefresh(server *testServer) func(*testing.T) {
 	}
 }
 
+// TestIsEnabledConcurrent verifies that concurrent calls to isEnabled on an MRF
+// sender don't race on the enabled field. isEnabled is invoked concurrently in
+// production from multiple producers calling sendPayloads on the same *sender.
+func TestIsEnabledConcurrent(t *testing.T) {
+	server := newTestServer()
+	defer server.Close()
+
+	s, err := newTestSender(server.URL)
+	assert.NoError(t, err)
+	defer s.Stop()
+
+	s.cfg.isMRF = true
+	failover := atomic.NewBool(false)
+	s.cfg.MRFFailoverAPM = failover.Load
+
+	var wg sync.WaitGroup
+	for i := range 50 {
+		wg.Go(func() {
+			failover.Store(i%2 == 0)
+			s.isEnabled()
+		})
+	}
+	wg.Wait()
+}
+
 func TestPayload(t *testing.T) {
 	expectBody := bytes.NewBufferString("body")
 	bodyLength := strconv.Itoa(expectBody.Len())