fix: data race on DCAClient.clusterAgentAPIClient in initHTTPClient (#54638)

### What does this PR do?

Fixes a data race on `DCAClient.clusterAgentAPIClient`: an unlocked bootstrap check raced with the field's other, locked accesses. Also guards against `startReconnectHandler()` being launched more than once concurrently.

### Motivation

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

### Describe how you validated your changes

Added a concurrent test on `initHTTPClient()`; fails under `-race` before this fix, passes after.


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

Author: pgimalac

pkg/util/clusteragent/clusteragent.go

@@ -104,6 +104,10 @@ type DCAClient struct {
 	clusterAgentVersion    version.Version // Version of the cluster-agent we're connected to
 	clusterAgentAPIClient  *http.Client
 	leaderClient           *leaderClient
+
+	// reconnectHandlerOnce ensures startReconnectHandler is only ever started once for
+	// this DCAClient, even if init() runs concurrently (e.g. via overlapping retries).
+	reconnectHandlerOnce sync.Once
 }
 
 // resetGlobalClusterAgentClient is a helper to remove the current DCAClient global
@@ -153,8 +157,12 @@ func (c *DCAClient) init() error {
 		return err
 	}
 
-	// Run DCA connection refresh
-	c.startReconnectHandler(time.Duration(pkgconfigsetup.Datadog().GetInt64("cluster_agent.client_reconnect_period_seconds")) * time.Second)
+	// Run DCA connection refresh. Guarded by a sync.Once so that concurrent calls to
+	// init() (e.g. overlapping retries from GetClusterAgentClient) don't spawn multiple
+	// reconnect-handler goroutines racing on this DCAClient's state.
+	c.reconnectHandlerOnce.Do(func() {
+		c.startReconnectHandler(time.Duration(pkgconfigsetup.Datadog().GetInt64("cluster_agent.client_reconnect_period_seconds")) * time.Second)
+	})
 
 	log.Infof("Successfully connected to the Datadog Cluster Agent %s", c.clusterAgentVersion.String())
 	return nil
@@ -208,9 +216,11 @@ func (c *DCAClient) initHTTPClient() error {
 	}
 
 	// We need to have a client to perform `GetVersion`, only happens during the first call
+	c.clusterAgentClientLock.Lock()
 	if c.clusterAgentAPIClient == nil {
 		c.clusterAgentAPIClient = clusterAgentAPIClient
 	}
+	c.clusterAgentClientLock.Unlock()
 
 	// Validate the cluster-agent client by checking the version
 	clusterAgentVersion, err := c.getVersion()

pkg/util/clusteragent/clusteragent_test.go

@@ -893,3 +893,54 @@ func (suite *clusterAgentSuite) TestDCAClientCertificateVerification() {
 		})
 	}
 }
+
+// TestInitHTTPClientConcurrent calls initHTTPClient() concurrently to catch data races on DCAClient state.
+func (suite *clusterAgentSuite) TestInitHTTPClientConcurrent() {
+	defer pkgapiutil.TestOnlyResetCrossNodeClientTLSConfig()
+	pkgapiutil.TestOnlyResetCrossNodeClientTLSConfig()
+
+	dca, err := newDummyClusterAgent(suite.config)
+	require.Nil(suite.T(), err, fmt.Sprintf("%v", err))
+
+	ts, p, err := dca.StartTLS()
+	require.Nil(suite.T(), err, fmt.Sprintf("%v", err))
+	defer ts.Close()
+
+	pkgapiutil.SetCrossNodeClientTLSConfig(&tls.Config{
+		InsecureSkipVerify: true,
+	})
+
+	c := &DCAClient{
+		clusterAgentAPIEndpoint: fmt.Sprintf("https://127.0.0.1:%d", p),
+	}
+	c.clusterAgentAPIRequestHeaders = http.Header{}
+	c.clusterAgentAPIRequestHeaders.Set(authorizationHeaderKey, "Bearer "+clusterAgentTokenValue)
+	c.clusterAgentAPIRequestHeaders.Set(RealIPHeader, clcRunnerIP)
+
+	// dummyClusterAgent.requests is a bounded channel that nothing else drains here;
+	// keep it flowing so ServeHTTP never blocks on a full channel and slows the test down.
+	stopDrain := make(chan struct{})
+	defer close(stopDrain)
+	go func() {
+		for {
+			select {
+			case <-dca.requests:
+			case <-stopDrain:
+				return
+			}
+		}
+	}()
+
+	const goroutines = 10
+	const iterations = 10
+
+	var wg sync.WaitGroup
+	for i := 0; i < goroutines; i++ {
+		wg.Go(func() {
+			for j := 0; j < iterations; j++ {
+				_ = c.initHTTPClient()
+			}
+		})
+	}
+	wg.Wait()
+}