fix: data race on StatsdClientWrapper.delegate in otel metrics client

SetDelegate correctly takes m.mutex, but all delegation methods (Gauge,
Count, Histogram, Incr, etc.) read m.delegate without the lock. Switch to
sync.RWMutex and take RLock in every delegation method to safely snapshot
the delegate before calling on it.

Found via a race-detector-enabled build in staging (see #54333 for context).

Author: pgimalac

comp/otelcol/otlp/components/metricsclient/metrics_client_test.go

@@ -247,3 +247,39 @@ func TestNoNilMeter(t *testing.T) {
 	_, err := InitializeMetricClient(&nilMeterProvider{}, ExporterSourceTag)
 	assert.ErrorIs(t, err, errNilMeter)
 }
+
+func TestStatsdClientWrapperConcurrentSetDelegateAndCount(t *testing.T) {
+	wrapper := NewStatsdClientWrapper(&statsd.NoOpClient{})
+
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				_ = wrapper.Count("test_count", 1, []string{"otlp:true"}, 1)
+			}
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			default:
+				wrapper.SetDelegate(&statsd.NoOpClient{})
+			}
+		}
+	}()
+
+	wg.Wait()
+}

comp/otelcol/otlp/components/metricsclient/metrics_client_wrapper.go

@@ -13,7 +13,7 @@ import (
 // StatsdClientWrapper is an implementation of ddgostatsd.ClientInterface that delegates all operations to the encompassed ddgostatsd.ClientInterface
 type StatsdClientWrapper struct {
 	delegate ddgostatsd.ClientInterface
-	mutex    sync.Mutex
+	mutex    sync.RWMutex
 }
 
 // NewStatsdClientWrapper returns a StatsdClientWrapper
@@ -33,95 +33,152 @@ var _ ddgostatsd.ClientInterface = (*StatsdClientWrapper)(nil)
 
 // Gauge measures the value of a metric at a particular time.
 func (m *StatsdClientWrapper) Gauge(name string, value float64, tags []string, rate float64) error {
-	return m.delegate.Gauge(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Gauge(name, value, tags, rate)
 }
 
 // GaugeWithTimestamp measures the value of a metric at a given time.
 func (m *StatsdClientWrapper) GaugeWithTimestamp(name string, value float64, tags []string, rate float64, timestamp time.Time) error {
-	return m.delegate.GaugeWithTimestamp(name, value, tags, rate, timestamp)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.GaugeWithTimestamp(name, value, tags, rate, timestamp)
 }
 
 // Count tracks how many times something happened per second.
 func (m *StatsdClientWrapper) Count(name string, value int64, tags []string, rate float64) error {
-	return m.delegate.Count(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Count(name, value, tags, rate)
 }
 
 // CountWithTimestamp tracks how many times something happened at the given second.
 func (m *StatsdClientWrapper) CountWithTimestamp(name string, value int64, tags []string, rate float64, timestamp time.Time) error {
-	return m.delegate.CountWithTimestamp(name, value, tags, rate, timestamp)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.CountWithTimestamp(name, value, tags, rate, timestamp)
 }
 
 // Histogram tracks the statistical distribution of a set of values on each host.
 func (m *StatsdClientWrapper) Histogram(name string, value float64, tags []string, rate float64) error {
-	return m.delegate.Histogram(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Histogram(name, value, tags, rate)
 }
 
 // Distribution tracks the statistical distribution of a set of values across your infrastructure.
 func (m *StatsdClientWrapper) Distribution(name string, value float64, tags []string, rate float64) error {
-	return m.delegate.Distribution(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Distribution(name, value, tags, rate)
 }
 
 // Decr is just Count of -1
 func (m *StatsdClientWrapper) Decr(name string, tags []string, rate float64) error {
-	return m.delegate.Decr(name, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Decr(name, tags, rate)
 }
 
 // Incr is just Count of 1
 func (m *StatsdClientWrapper) Incr(name string, tags []string, rate float64) error {
-	return m.delegate.Incr(name, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Incr(name, tags, rate)
 }
 
 // Set counts the number of unique elements in a group.
 func (m *StatsdClientWrapper) Set(name string, value string, tags []string, rate float64) error {
-	return m.delegate.Set(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Set(name, value, tags, rate)
 }
 
 // Timing sends timing information, it is an alias for TimeInMilliseconds
 func (m *StatsdClientWrapper) Timing(name string, value time.Duration, tags []string, rate float64) error {
-	return m.delegate.Timing(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Timing(name, value, tags, rate)
 }
 
 // TimeInMilliseconds sends timing information in milliseconds.
 func (m *StatsdClientWrapper) TimeInMilliseconds(name string, value float64, tags []string, rate float64) error {
-	return m.delegate.TimeInMilliseconds(name, value, tags, rate)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.TimeInMilliseconds(name, value, tags, rate)
 }
 
 // Event sends the provided Event.
 func (m *StatsdClientWrapper) Event(e *ddgostatsd.Event) error {
-	return m.delegate.Event(e)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Event(e)
 }
 
 // SimpleEvent sends an event with the provided title and text.
 func (m *StatsdClientWrapper) SimpleEvent(title, text string) error {
-	return m.delegate.SimpleEvent(title, text)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.SimpleEvent(title, text)
 }
 
 // ServiceCheck sends the provided ServiceCheck.
 func (m *StatsdClientWrapper) ServiceCheck(sc *ddgostatsd.ServiceCheck) error {
-	return m.delegate.ServiceCheck(sc)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.ServiceCheck(sc)
 }
 
 // SimpleServiceCheck sends an serviceCheck with the provided name and status.
 func (m *StatsdClientWrapper) SimpleServiceCheck(name string, status ddgostatsd.ServiceCheckStatus) error {
-	return m.delegate.SimpleServiceCheck(name, status)
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.SimpleServiceCheck(name, status)
 }
 
 // Close the client connection.
 func (m *StatsdClientWrapper) Close() error {
-	return m.delegate.Close()
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Close()
 }
 
 // Flush forces a flush of all the queued dogstatsd payloads.
 func (m *StatsdClientWrapper) Flush() error {
-	return m.delegate.Flush()
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.Flush()
 }
 
 // IsClosed returns if the client has been closed.
 func (m *StatsdClientWrapper) IsClosed() bool {
-	return m.delegate.IsClosed()
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.IsClosed()
 }
 
 // GetTelemetry return the telemetry metrics for the client since it started.
 func (m *StatsdClientWrapper) GetTelemetry() ddgostatsd.Telemetry {
-	return m.delegate.GetTelemetry()
+	m.mutex.RLock()
+	delegate := m.delegate
+	m.mutex.RUnlock()
+	return delegate.GetTelemetry()
 }