Serialize access to RemotelyControlledSampler.sampler (#515)

Serialize read access for RemotelyControlledSampler.sampler field

Signed-off-by: Dima Kozlov <hummerd@mail.ru>

sampler_remote.go

@@ -64,7 +64,7 @@ type RemotelyControlledSampler struct {
 	// Cf. https://github.com/uber/jaeger-client-go/issues/155, https://goo.gl/zW7dgq
 	closed int64 // 0 - not closed, 1 - closed
 
-	sync.RWMutex
+	sync.RWMutex // used to serialize access to samplerOptions.sampler
 	samplerOptions
 
 	serviceName string
@@ -95,22 +95,22 @@ func (s *RemotelyControlledSampler) IsSampled(id TraceID, operation string) (boo
 
 // OnCreateSpan implements OnCreateSpan of SamplerV2.
 func (s *RemotelyControlledSampler) OnCreateSpan(span *Span) SamplingDecision {
-	return s.sampler.OnCreateSpan(span)
+	return s.Sampler().OnCreateSpan(span)
 }
 
 // OnSetOperationName implements OnSetOperationName of SamplerV2.
 func (s *RemotelyControlledSampler) OnSetOperationName(span *Span, operationName string) SamplingDecision {
-	return s.sampler.OnSetOperationName(span, operationName)
+	return s.Sampler().OnSetOperationName(span, operationName)
 }
 
 // OnSetTag implements OnSetTag of SamplerV2.
 func (s *RemotelyControlledSampler) OnSetTag(span *Span, key string, value interface{}) SamplingDecision {
-	return s.sampler.OnSetTag(span, key, value)
+	return s.Sampler().OnSetTag(span, key, value)
 }
 
 // OnFinishSpan implements OnFinishSpan of SamplerV2.
 func (s *RemotelyControlledSampler) OnFinishSpan(span *Span) SamplingDecision {
-	return s.sampler.OnFinishSpan(span)
+	return s.Sampler().OnFinishSpan(span)
 }
 
 // Close implements Close() of Sampler.
@@ -153,8 +153,8 @@ func (s *RemotelyControlledSampler) pollControllerWithTicker(ticker *time.Ticker
 
 // Sampler returns the currently active sampler.
 func (s *RemotelyControlledSampler) Sampler() SamplerV2 {
-	s.Lock()
-	defer s.Unlock()
+	s.RLock()
+	defer s.RUnlock()
 	return s.sampler
 }
 

sampler_remote_test.go

@@ -201,7 +201,7 @@ func TestRemotelyControlledSampler_updateSampler(t *testing.T) {
 			agent, sampler, metricsFactory := initAgent(t)
 			defer agent.Close()
 
-			initSampler, ok := sampler.sampler.(*ProbabilisticSampler)
+			initSampler, ok := sampler.Sampler().(*ProbabilisticSampler)
 			assert.True(t, ok)
 
 			res := &sampling.SamplingStrategyResponse{
@@ -231,9 +231,9 @@ func TestRemotelyControlledSampler_updateSampler(t *testing.T) {
 				},
 			)
 
-			s, ok := sampler.sampler.(*PerOperationSampler)
+			s, ok := sampler.Sampler().(*PerOperationSampler)
 			assert.True(t, ok)
-			assert.NotEqual(t, initSampler, sampler.sampler, "Sampler should have been updated")
+			assert.NotEqual(t, initSampler, sampler.Sampler(), "Sampler should have been updated")
 			assert.Equal(t, test.expectedDefaultProbability, s.defaultSampler.SamplingRate())
 
 			// First call is always sampled
@@ -306,13 +306,13 @@ func TestSamplerQueryError(t *testing.T) {
 	// override the actual handler
 	sampler.samplingFetcher = &fakeSamplingFetcher{}
 
-	initSampler, ok := sampler.sampler.(*ProbabilisticSampler)
+	initSampler, ok := sampler.Sampler().(*ProbabilisticSampler)
 	assert.True(t, ok)
 
 	sampler.Close() // stop timer-based updates, we want to call them manually
 
 	sampler.UpdateSampler()
-	assert.Equal(t, initSampler, sampler.sampler, "Sampler should not have been updated due to query error")
+	assert.Equal(t, initSampler, sampler.Sampler(), "Sampler should not have been updated due to query error")
 
 	metricsFactory.AssertCounterMetrics(t,
 		mTestutils.ExpectedMetric{Name: "jaeger.tracer.sampler_queries", Tags: map[string]string{"result": "err"}, Value: 1},
@@ -340,29 +340,29 @@ func TestRemotelyControlledSampler_updateSamplerFromAdaptiveSampler(t *testing.T
 	})
 
 	// Overwrite the sampler with an adaptive sampler
-	remoteSampler.sampler = adaptiveSampler
+	remoteSampler.setSampler(adaptiveSampler)
 
 	agent.AddSamplingStrategy("client app",
 		getSamplingStrategyResponse(sampling.SamplingStrategyType_PROBABILISTIC, 0.5))
 	remoteSampler.UpdateSampler()
 
 	// Sampler should have been updated to probabilistic
-	_, ok := remoteSampler.sampler.(*ProbabilisticSampler)
+	_, ok := remoteSampler.Sampler().(*ProbabilisticSampler)
 	require.True(t, ok)
 
 	// Overwrite the sampler with an adaptive sampler
-	remoteSampler.sampler = adaptiveSampler
+	remoteSampler.setSampler(adaptiveSampler)
 
 	agent.AddSamplingStrategy("client app",
 		getSamplingStrategyResponse(sampling.SamplingStrategyType_RATE_LIMITING, 1))
 	remoteSampler.UpdateSampler()
 
 	// Sampler should have been updated to ratelimiting
-	_, ok = remoteSampler.sampler.(*RateLimitingSampler)
+	_, ok = remoteSampler.Sampler().(*RateLimitingSampler)
 	require.True(t, ok)
 
 	// Overwrite the sampler with an adaptive sampler
-	remoteSampler.sampler = adaptiveSampler
+	remoteSampler.setSampler(adaptiveSampler)
 
 	// Update existing adaptive sampler
 	agent.AddSamplingStrategy("client app", &sampling.SamplingStrategyResponse{OperationSampling: strategies})
@@ -460,15 +460,15 @@ func TestRemotelyControlledSampler_updateRateLimitingOrProbabilisticSampler(t *t
 				return
 			}
 			if testCase.referenceEquivalence {
-				assert.Equal(t, testCase.expectedSampler, remoteSampler.sampler)
+				assert.Equal(t, testCase.expectedSampler, remoteSampler.Sampler())
 			} else {
 				type comparable interface {
 					Equal(other Sampler) bool
 				}
 				es, esOk := testCase.expectedSampler.(comparable)
 				require.True(t, esOk, "expected sampler %+v must implement Equal()", testCase.expectedSampler)
-				assert.True(t, es.Equal(remoteSampler.sampler.(Sampler)),
-					"sampler.Equal: want=%+v, have=%+v", testCase.expectedSampler, remoteSampler.sampler)
+				assert.True(t, es.Equal(remoteSampler.Sampler().(Sampler)),
+					"sampler.Equal: want=%+v, have=%+v", testCase.expectedSampler, remoteSampler.Sampler())
 			}
 		})
 	}

sampler_test.go

@@ -53,7 +53,7 @@ func TestSamplerTags(t *testing.T) {
 	require.NoError(t, err)
 	rate := NewRateLimitingSampler(0.1)
 	remote := &RemotelyControlledSampler{}
-	remote.sampler = NewConstSampler(true)
+	remote.setSampler(NewConstSampler(true))
 	tests := []struct {
 		sampler  SamplerV2
 		typeTag  string
@@ -395,7 +395,7 @@ func TestAdaptiveSampler_lockRaceCondition(t *testing.T) {
 		},
 	})
 	// Overwrite the sampler with an adaptive sampler
-	remoteSampler.sampler = adaptiveSampler
+	remoteSampler.setSampler(adaptiveSampler)
 
 	tracer, closer := NewTracer("service", remoteSampler, NewNullReporter())
 	defer closer.Close()