fix(otelcol/metricsclient): fix data race on StatsdClientWrapper.delegate

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

@@ -247,3 +247,34 @@ func TestNoNilMeter(t *testing.T) {
 	_, err := InitializeMetricClient(&nilMeterProvider{}, ExporterSourceTag)
 	assert.ErrorIs(t, err, errNilMeter)
 }
+
+func TestStatsdClientWrapperConcurrentSetDelegateAndCount(t *testing.T) {
+	t.Parallel()
+
+	wrapper := NewStatsdClientWrapper(&statsd.NoOpClient{})
+
+	const iterations = 5000
+	var wg sync.WaitGroup
+	var start sync.WaitGroup
+
+	// Start barrier so both goroutines begin looping at the same time,
+	// maximising overlap for the race detector without a wall-clock delay.
+	start.Add(1)
+
+	wg.Go(func() {
+		start.Wait()
+		for range iterations {
+			_ = wrapper.Count("test_count", 1, []string{"otlp:true"}, 1)
+		}
+	})
+
+	wg.Go(func() {
+		start.Wait()
+		for range iterations {
+			wrapper.SetDelegate(&statsd.NoOpClient{})
+		}
+	})
+
+	start.Done() // release both goroutines
+	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
@@ -29,99 +29,108 @@ func (m *StatsdClientWrapper) SetDelegate(cl ddgostatsd.ClientInterface) {
 	m.delegate = cl
 }
 
+// getDelegate returns a snapshot of the current delegate under the read lock.
+// Callers must not mutate the returned interface; it is used only to dispatch
+// the current operation on the delegate that was live at the time of the call.
+func (m *StatsdClientWrapper) getDelegate() ddgostatsd.ClientInterface {
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+	return m.delegate
+}
+
 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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().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)
+	return m.getDelegate().TimeInMilliseconds(name, value, tags, rate)
 }
 
 // Event sends the provided Event.
 func (m *StatsdClientWrapper) Event(e *ddgostatsd.Event) error {
-	return m.delegate.Event(e)
+	return m.getDelegate().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)
+	return m.getDelegate().SimpleEvent(title, text)
 }
 
 // ServiceCheck sends the provided ServiceCheck.
 func (m *StatsdClientWrapper) ServiceCheck(sc *ddgostatsd.ServiceCheck) error {
-	return m.delegate.ServiceCheck(sc)
+	return m.getDelegate().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)
+	return m.getDelegate().SimpleServiceCheck(name, status)
 }
 
 // Close the client connection.
 func (m *StatsdClientWrapper) Close() error {
-	return m.delegate.Close()
+	return m.getDelegate().Close()
 }
 
 // Flush forces a flush of all the queued dogstatsd payloads.
 func (m *StatsdClientWrapper) Flush() error {
-	return m.delegate.Flush()
+	return m.getDelegate().Flush()
 }
 
 // IsClosed returns if the client has been closed.
 func (m *StatsdClientWrapper) IsClosed() bool {
-	return m.delegate.IsClosed()
+	return m.getDelegate().IsClosed()
 }
 
 // GetTelemetry return the telemetry metrics for the client since it started.
 func (m *StatsdClientWrapper) GetTelemetry() ddgostatsd.Telemetry {
-	return m.delegate.GetTelemetry()
+	return m.getDelegate().GetTelemetry()
 }