Merge pull request #406 from axw/tracer-active-race

apm: fix data race accessing Tracer.active

CHANGELOG.md

@@ -1,6 +1,10 @@
 # Changelog
 
-## [Unreleased](https://github.com/elastic/apm-agent-go/compare/v1.1.1...master)
+## [Unreleased](https://github.com/elastic/apm-agent-go/compare/v1.1.2...master)
+
+## [v1.1.2](https://github.com/elastic/apm-agent-go/releases/tag/v1.1.2)
+
+ - Fix data race between Tracer.Active and Tracer.loop (#406)
 
 ## [v1.1.1](https://github.com/elastic/apm-agent-go/releases/tag/v1.1.1)
 

env_test.go

@@ -270,7 +270,7 @@ func TestTracerSpanFramesMinDurationEnvInvalid(t *testing.T) {
 	assert.EqualError(t, err, "failed to parse ELASTIC_APM_SPAN_FRAMES_MIN_DURATION: invalid duration aeon")
 }
 
-func TestTracerActive(t *testing.T) {
+func TestTracerActiveEnv(t *testing.T) {
 	os.Setenv("ELASTIC_APM_ACTIVE", "false")
 	defer os.Unsetenv("ELASTIC_APM_ACTIVE")
 
@@ -285,7 +285,7 @@ func TestTracerActive(t *testing.T) {
 	assert.Zero(t, transport.Payloads())
 }
 
-func TestTracerActiveInvalid(t *testing.T) {
+func TestTracerActiveEnvInvalid(t *testing.T) {
 	os.Setenv("ELASTIC_APM_ACTIVE", "yep")
 	defer os.Unsetenv("ELASTIC_APM_ACTIVE")
 

tracer.go

@@ -8,6 +8,7 @@ import (
 	"log"
 	"math/rand"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"go.elastic.co/apm/internal/apmconfig"
@@ -180,7 +181,7 @@ type Tracer struct {
 	process *model.Process
 	system  *model.System
 
-	active            bool
+	active            int32
 	bufferSize        int
 	metricsBufferSize int
 	closing           chan struct{}
@@ -247,19 +248,20 @@ func newTracer(opts options) *Tracer {
 		transactions:          make(chan *TransactionData, transactionsChannelCap),
 		spans:                 make(chan *SpanData, spansChannelCap),
 		errors:                make(chan *ErrorData, errorsChannelCap),
+		active:                1,
 		maxSpans:              opts.maxSpans,
 		sampler:               opts.sampler,
 		captureBody:           opts.captureBody,
 		spanFramesMinDuration: opts.spanFramesMinDuration,
-		active:                opts.active,
 		bufferSize:            opts.bufferSize,
 		metricsBufferSize:     opts.metricsBufferSize,
 	}
 	t.Service.Name = opts.serviceName
 	t.Service.Version = opts.serviceVersion
 	t.Service.Environment = opts.serviceEnvironment
 
-	if !t.active {
+	if !opts.active {
+		t.active = 0
 		close(t.closed)
 		return t
 	}
@@ -325,7 +327,7 @@ func (t *Tracer) Flush(abort <-chan struct{}) {
 // Active reports whether the tracer is active. If the tracer is inactive,
 // no transactions or errors will be sent to the Elastic APM server.
 func (t *Tracer) Active() bool {
-	return t.active
+	return atomic.LoadInt32(&t.active) == 1
 }
 
 // SetRequestDuration sets the maximum amount of time to keep a request open
@@ -481,7 +483,7 @@ func (t *Tracer) loop() {
 	ctx, cancelContext := context.WithCancel(context.Background())
 	defer cancelContext()
 	defer close(t.closed)
-	defer func() { t.active = false }()
+	defer atomic.StoreInt32(&t.active, 0)
 
 	var req iochan.ReadRequest
 	var requestBuf bytes.Buffer

tracer_test.go

@@ -411,6 +411,21 @@ func TestTracerKubernetesMetadata(t *testing.T) {
 	})
 }
 
+func TestTracerActive(t *testing.T) {
+	tracer, _ := transporttest.NewRecorderTracer()
+	defer tracer.Close()
+	assert.True(t, tracer.Active())
+
+	// Kick off calls to tracer.Active concurrently
+	// with the tracer.Close, to test that we ensure
+	// there are no data races.
+	go func() {
+		for i := 0; i < 100; i++ {
+			tracer.Active()
+		}
+	}()
+}
+
 type blockedTransport struct {
 	transport.Transport
 	unblocked chan struct{}

version.go

@@ -2,5 +2,5 @@ package apm
 
 const (
 	// AgentVersion is the Elastic APM Go Agent version.
-	AgentVersion = "1.1.1"
+	AgentVersion = "1.1.2"
 )