diff --git a/comp/otelcol/otlp/components/metricsclient/metrics_client_test.go b/comp/otelcol/otlp/components/metricsclient/metrics_client_test.go index 30e490265971..9edc630e1563 100644 --- a/comp/otelcol/otlp/components/metricsclient/metrics_client_test.go +++ b/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() +} diff --git a/comp/otelcol/otlp/components/metricsclient/metrics_client_wrapper.go b/comp/otelcol/otlp/components/metricsclient/metrics_client_wrapper.go index 635c773ce3e9..9eb9b2d0bd60 100644 --- a/comp/otelcol/otlp/components/metricsclient/metrics_client_wrapper.go +++ b/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() }