fix: indefinite function runs inside mutex lock

stats/statsd.go

@@ -50,17 +50,28 @@
 	goFactory.Go(func() {
 		if err != nil {
 			s.logger.Info("retrying StatsD client creation in the background...")
-			s.state.client.statsdMu.Lock()
-			s.state.client.statsd, err = s.getNewStatsdClientWithExpoBackoff(ctx, s.state.conn, s.statsdConfig.statsdTagsFormat(), s.statsdConfig.statsdDefaultTags())
-			s.state.client.statsdMu.Unlock()
+			var c *statsd.Client
+			c, err = s.getNewStatsdClientWithExpoBackoff(
+				ctx,
+				s.state.conn,
+				s.statsdConfig.statsdTagsFormat(),
+				s.statsdConfig.statsdDefaultTags(),
+			)
 			if err != nil {
 				s.config.enabled.Store(false)
 				s.logger.Errorf("error while creating new StatsD client, giving up: %v", err)
 			} else {
 				s.state.clientsLock.Lock()
+				s.state.client.statsd = c
 				for _, client := range s.state.pendingClients {
 					client.statsdMu.Lock()
-					client.statsd = s.state.client.statsd.Clone(s.state.conn, s.statsdConfig.statsdTagsFormat(), s.statsdConfig.statsdDefaultTags(), statsd.Tags(client.tags...), statsd.SampleRate(client.samplingRate))
+					client.statsd = s.state.client.statsd.Clone(
+						s.state.conn,
+						s.statsdConfig.statsdTagsFormat(),
+						s.statsdConfig.statsdDefaultTags(),
+						statsd.Tags(client.tags...),
+						statsd.SampleRate(client.samplingRate),
+					)
 					client.statsdMu.Unlock()
 				}
 
@@ -104,7 +115,7 @@
 
 func (s *statsdStats) collectPeriodicStats(goFactory GoRoutineFactory) {
 	gaugeFunc := func(key string, val uint64) {
-		s.state.client.statsd.Gauge("runtime_"+key, val)
+		s.NewStat("runtime_"+key, GaugeType).Gauge(val)
 	}
 	s.state.rc = newRuntimeStatsCollector(gaugeFunc)
 	s.state.rc.PauseDur = time.Duration(s.config.periodicStatsConfig.statsCollectionInterval) * time.Second
@@ -152,7 +163,7 @@
 
 // NewStat creates a new Measurement with provided Name and Type
 func (s *statsdStats) NewStat(name, statType string) (m Measurement) {
-	return s.newStatsdMeasurement(name, statType, s.state.client)
+	return s.internalNewTaggedStat(name, statType, nil, 1)
 }
 
 func (s *statsdStats) NewTaggedStat(Name, StatType string, tags Tags) (m Measurement) {
@@ -200,7 +211,13 @@
 			taggedClient = &statsdClient{samplingRate: samplingRate, tags: tagVals}
 			if s.state.connEstablished {
 				taggedClient.statsdMu.Lock()
-				taggedClient.statsd = s.state.client.statsd.Clone(s.state.conn, s.statsdConfig.statsdTagsFormat(), s.statsdConfig.statsdDefaultTags(), statsd.Tags(tagVals...), statsd.SampleRate(samplingRate))
+				taggedClient.statsd = s.state.client.statsd.Clone(
+					s.state.conn,
+					s.statsdConfig.statsdTagsFormat(),
+					s.statsdConfig.statsdDefaultTags(),
+					statsd.Tags(tagVals...),
+					statsd.SampleRate(samplingRate),
+				)
 				taggedClient.statsdMu.Unlock()
 			} else {
 				// new statsd clients will be created when connection is established for all pending clients
@@ -297,9 +314,8 @@
 }
 
 // ready returns true if the statsd client is ready to be used (not nil).
+//
+// statsdMu.RLock should be held when calling this method.
 func (sc *statsdClient) ready() bool {
-	sc.statsdMu.RLock()
-	defer sc.statsdMu.RUnlock()
-
 	return sc.statsd != nil
 }

stats/statsd_measurement.go

@@ -25,6 +25,8 @@
 }
 
 func (c *statsdCounter) Count(n int) {
+	c.client.statsdMu.RLock()
+	defer c.client.statsdMu.RUnlock()
 	if c.skip() {
 		return
 	}
@@ -33,6 +35,8 @@
 
 // Increment increases the stat by 1. Is the Equivalent of Count(1). Only applies to CountType stats
 func (c *statsdCounter) Increment() {
+	c.client.statsdMu.RLock()
+	defer c.client.statsdMu.RUnlock()
 	if c.skip() {
 		return
 	}
@@ -46,6 +50,8 @@
 
 // Gauge records an absolute value for this stat. Only applies to GaugeType stats
 func (g *statsdGauge) Gauge(value interface{}) {
+	g.client.statsdMu.RLock()
+	defer g.client.statsdMu.RUnlock()
 	if g.skip() {
 		return
 	}
@@ -61,6 +67,8 @@
 // Start starts a new timing for this stat. Only applies to TimerType stats
 // Deprecated: Use concurrent safe SendTiming() instead
 func (t *statsdTimer) Start() {
+	t.client.statsdMu.RLock()
+	defer t.client.statsdMu.RUnlock()
 	if t.skip() {
 		return
 	}
@@ -71,6 +79,8 @@
 // End send the time elapsed since the Start()  call of this stat. Only applies to TimerType stats
 // Deprecated: Use concurrent safe SendTiming() instead
 func (t *statsdTimer) End() {
+	t.client.statsdMu.RLock()
+	defer t.client.statsdMu.RUnlock()
 	if t.skip() || t.timing == nil {
 		return
 	}
@@ -84,6 +94,8 @@
 
 // SendTiming sends a timing for this stat. Only applies to TimerType stats
 func (t *statsdTimer) SendTiming(duration time.Duration) {
+	t.client.statsdMu.RLock()
+	defer t.client.statsdMu.RUnlock()
 	if t.skip() {
 		return
 	}
@@ -107,6 +119,8 @@
 
 // Observe sends an observation
 func (h *statsdHistogram) Observe(value float64) {
+	h.client.statsdMu.RLock()
+	defer h.client.statsdMu.RUnlock()
 	if h.skip() {
 		return
 	}