tracer.go

// Licensed to Elasticsearch B.V. under one or more contributor
// license agreements. See the NOTICE file distributed with
// this work for additional information regarding copyright
// ownership. Elasticsearch B.V. licenses this file to you under
// the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package apm

import (
	"bytes"
	"compress/zlib"
	"context"
	"io"
	"log"
	"math/rand"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"go.elastic.co/apm/apmconfig"
	"go.elastic.co/apm/internal/apmlog"
	"go.elastic.co/apm/internal/configutil"
	"go.elastic.co/apm/internal/iochan"
	"go.elastic.co/apm/internal/ringbuffer"
	"go.elastic.co/apm/internal/wildcard"
	"go.elastic.co/apm/model"
	"go.elastic.co/apm/stacktrace"
	"go.elastic.co/apm/transport"
	"go.elastic.co/fastjson"
)

const (
	defaultPreContext     = 3
	defaultPostContext    = 3
	gracePeriodJitter     = 0.1 // +/- 10%
	tracerEventChannelCap = 1000
)

var (
	// DefaultTracer is the default global Tracer, set at package
	// initialization time, configured via environment variables.
	//
	// This will always be initialized to a non-nil value. If any
	// of the environment variables are invalid, the corresponding
	// errors will be logged to stderr and the default values will
	// be used instead.
	DefaultTracer *Tracer
)

func init() {
	var opts TracerOptions
	opts.initDefaults(true)
	DefaultTracer = newTracer(opts)
}

// TracerOptions holds initial tracer options, for passing to NewTracerOptions.
type TracerOptions struct {
	// ServiceName holds the service name.
	//
	// If ServiceName is empty, the service name will be defined using the
	// ELASTIC_APM_SERVICE_NAME environment variable, or if that is not set,
	// the executable name.
	ServiceName string

	// ServiceVersion holds the service version.
	//
	// If ServiceVersion is empty, the service version will be defined using
	// the ELASTIC_APM_SERVICE_VERSION environment variable.
	ServiceVersion string

	// ServiceEnvironment holds the service environment.
	//
	// If ServiceEnvironment is empty, the service environment will be defined
	// using the ELASTIC_APM_ENVIRONMENT environment variable.
	ServiceEnvironment string

	// Transport holds the transport to use for sending events.
	//
	// If Transport is nil, transport.Default will be used.
	//
	// If Transport implements apmconfig.Watcher, the tracer will begin watching
	// for remote changes immediately. This behaviour can be disabled by setting
	// the environment variable ELASTIC_APM_CENTRAL_CONFIG=false.
	Transport transport.Transport

	requestDuration       time.Duration
	metricsInterval       time.Duration
	maxSpans              int
	requestSize           int
	bufferSize            int
	metricsBufferSize     int
	sampler               Sampler
	sanitizedFieldNames   wildcard.Matchers
	disabledMetrics       wildcard.Matchers
	captureHeaders        bool
	captureBody           CaptureBodyMode
	spanFramesMinDuration time.Duration
	stackTraceLimit       int
	active                bool
	configWatcher         apmconfig.Watcher
	breakdownMetrics      bool
}

// initDefaults updates opts with default values.
func (opts *TracerOptions) initDefaults(continueOnError bool) error {
	var errs []error
	failed := func(err error) bool {
		if err == nil {
			return false
		}
		errs = append(errs, err)
		return true
	}

	requestDuration, err := initialRequestDuration()
	if failed(err) {
		requestDuration = defaultAPIRequestTime
	}

	metricsInterval, err := initialMetricsInterval()
	if err != nil {
		metricsInterval = defaultMetricsInterval
		errs = append(errs, err)
	}

	requestSize, err := initialAPIRequestSize()
	if err != nil {
		requestSize = int(defaultAPIRequestSize)
		errs = append(errs, err)
	}

	bufferSize, err := initialAPIBufferSize()
	if err != nil {
		bufferSize = int(defaultAPIBufferSize)
		errs = append(errs, err)
	}

	metricsBufferSize, err := initialMetricsBufferSize()
	if err != nil {
		metricsBufferSize = int(defaultMetricsBufferSize)
		errs = append(errs, err)
	}

	maxSpans, err := initialMaxSpans()
	if failed(err) {
		maxSpans = defaultMaxSpans
	}

	sampler, err := initialSampler()
	if failed(err) {
		sampler = nil
	}

	captureHeaders, err := initialCaptureHeaders()
	if failed(err) {
		captureHeaders = defaultCaptureHeaders
	}

	captureBody, err := initialCaptureBody()
	if failed(err) {
		captureBody = CaptureBodyOff
	}

	spanFramesMinDuration, err := initialSpanFramesMinDuration()
	if failed(err) {
		spanFramesMinDuration = defaultSpanFramesMinDuration
	}

	stackTraceLimit, err := initialStackTraceLimit()
	if failed(err) {
		stackTraceLimit = defaultStackTraceLimit
	}

	active, err := initialActive()
	if failed(err) {
		active = true
	}

	centralConfigEnabled, err := initialCentralConfigEnabled()
	if failed(err) {
		centralConfigEnabled = true
	}

	breakdownMetricsEnabled, err := initialBreakdownMetricsEnabled()
	if failed(err) {
		breakdownMetricsEnabled = true
	}

	if opts.ServiceName != "" {
		err := validateServiceName(opts.ServiceName)
		if failed(err) {
			opts.ServiceName = ""
		}
	}

	if len(errs) != 0 && !continueOnError {
		return errs[0]
	}
	for _, err := range errs {
		log.Printf("[apm]: %s", err)
	}

	opts.requestDuration = requestDuration
	opts.metricsInterval = metricsInterval
	opts.requestSize = requestSize
	opts.bufferSize = bufferSize
	opts.metricsBufferSize = metricsBufferSize
	opts.maxSpans = maxSpans
	opts.sampler = sampler
	opts.sanitizedFieldNames = initialSanitizedFieldNames()
	opts.disabledMetrics = initialDisabledMetrics()
	opts.breakdownMetrics = breakdownMetricsEnabled
	opts.captureHeaders = captureHeaders
	opts.captureBody = captureBody
	opts.spanFramesMinDuration = spanFramesMinDuration
	opts.stackTraceLimit = stackTraceLimit
	opts.active = active
	if opts.Transport == nil {
		opts.Transport = transport.Default
	}
	if centralConfigEnabled {
		if cw, ok := opts.Transport.(apmconfig.Watcher); ok {
			opts.configWatcher = cw
		}
	}

	serviceName, serviceVersion, serviceEnvironment := initialService()
	if opts.ServiceName == "" {
		opts.ServiceName = serviceName
	}
	if opts.ServiceVersion == "" {
		opts.ServiceVersion = serviceVersion
	}
	if opts.ServiceEnvironment == "" {
		opts.ServiceEnvironment = serviceEnvironment
	}
	return nil
}

// Tracer manages the sampling and sending of transactions to
// Elastic APM.
//
// Transactions are buffered until they are flushed (forcibly
// with a Flush call, or when the flush timer expires), or when
// the maximum transaction queue size is reached. Failure to
// send will be periodically retried. Once the queue limit has
// been reached, new transactions will replace older ones in
// the queue.
//
// Errors are sent as soon as possible, but will buffered and
// later sent in bulk if the tracer is busy, or otherwise cannot
// send to the server, e.g. due to network failure. There is
// a limit to the number of errors that will be buffered, and
// once that limit has been reached, new errors will be dropped
// until the queue is drained.
//
// The exported fields be altered or replaced any time up until
// any Tracer methods have been invoked.
type Tracer struct {
	Transport transport.Transport
	Service   struct {
		Name        string
		Version     string
		Environment string
	}

	process *model.Process
	system  *model.System

	active            int32
	bufferSize        int
	metricsBufferSize int
	closing           chan struct{}
	closed            chan struct{}
	forceFlush        chan chan<- struct{}
	forceSendMetrics  chan chan<- struct{}
	configCommands    chan tracerConfigCommand
	configWatcher     chan apmconfig.Watcher
	events            chan tracerEvent
	breakdownMetrics  *breakdownMetrics

	statsMu sync.Mutex
	stats   TracerStats

	maxSpansMu sync.RWMutex
	maxSpans   int

	spanFramesMinDurationMu sync.RWMutex
	spanFramesMinDuration   time.Duration

	stackTraceLimitMu sync.RWMutex
	stackTraceLimit   int

	samplerMu sync.RWMutex
	sampler   Sampler
	// localSampler holds the most recently locally-configured Sampler,
	// which is what sampler will be set to when a centrally defined
	// sampler is removed.
	localSampler Sampler
	// remoteSampler records whether sampler is defined by the remote
	// config watcher.
	remoteSampler bool

	captureHeadersMu sync.RWMutex
	captureHeaders   bool

	captureBodyMu sync.RWMutex
	captureBody   CaptureBodyMode

	errorDataPool       sync.Pool
	spanDataPool        sync.Pool
	transactionDataPool sync.Pool
}

// NewTracer returns a new Tracer, using the default transport,
// and with the specified service name and version if specified.
// This is equivalent to calling NewTracerOptions with a
// TracerOptions having ServiceName and ServiceVersion set to
// the provided arguments.
func NewTracer(serviceName, serviceVersion string) (*Tracer, error) {
	return NewTracerOptions(TracerOptions{
		ServiceName:    serviceName,
		ServiceVersion: serviceVersion,
	})
}

// NewTracerOptions returns a new Tracer using the provided options.
// See TracerOptions for details on the options, and their default
// values.
func NewTracerOptions(opts TracerOptions) (*Tracer, error) {
	if err := opts.initDefaults(false); err != nil {
		return nil, err
	}
	return newTracer(opts), nil
}

func newTracer(opts TracerOptions) *Tracer {
	t := &Tracer{
		Transport:             opts.Transport,
		process:               &currentProcess,
		system:                &localSystem,
		closing:               make(chan struct{}),
		closed:                make(chan struct{}),
		forceFlush:            make(chan chan<- struct{}),
		forceSendMetrics:      make(chan chan<- struct{}),
		configCommands:        make(chan tracerConfigCommand),
		configWatcher:         make(chan apmconfig.Watcher),
		events:                make(chan tracerEvent, tracerEventChannelCap),
		active:                1,
		breakdownMetrics:      newBreakdownMetrics(),
		maxSpans:              opts.maxSpans,
		sampler:               opts.sampler,
		localSampler:          opts.sampler,
		captureHeaders:        opts.captureHeaders,
		captureBody:           opts.captureBody,
		spanFramesMinDuration: opts.spanFramesMinDuration,
		stackTraceLimit:       opts.stackTraceLimit,
		bufferSize:            opts.bufferSize,
		metricsBufferSize:     opts.metricsBufferSize,
	}
	t.Service.Name = opts.ServiceName
	t.Service.Version = opts.ServiceVersion
	t.Service.Environment = opts.ServiceEnvironment
	t.breakdownMetrics.enabled = opts.breakdownMetrics

	if !opts.active {
		t.active = 0
		close(t.closed)
		return t
	}

	go t.loop()
	t.configCommands <- func(cfg *tracerConfig) {
		cfg.metricsInterval = opts.metricsInterval
		cfg.requestDuration = opts.requestDuration
		cfg.requestSize = opts.requestSize
		cfg.sanitizedFieldNames = opts.sanitizedFieldNames
		cfg.disabledMetrics = opts.disabledMetrics
		cfg.preContext = defaultPreContext
		cfg.postContext = defaultPostContext
		cfg.metricsGatherers = []MetricsGatherer{newBuiltinMetricsGatherer(t)}
		if apmlog.DefaultLogger != nil {
			cfg.logger = apmlog.DefaultLogger
		}
	}
	if opts.configWatcher != nil {
		t.configWatcher <- opts.configWatcher
	}
	return t
}

// tracerConfig holds the tracer's runtime configuration, which may be modified
// by sending a tracerConfigCommand to the tracer's configCommands channel.
type tracerConfig struct {
	requestSize             int
	requestDuration         time.Duration
	metricsInterval         time.Duration
	logger                  WarningLogger
	metricsGatherers        []MetricsGatherer
	contextSetter           stacktrace.ContextSetter
	preContext, postContext int
	sanitizedFieldNames     wildcard.Matchers
	disabledMetrics         wildcard.Matchers
}

type tracerConfigCommand func(*tracerConfig)

// Close closes the Tracer, preventing transactions from being
// sent to the APM server.
func (t *Tracer) Close() {
	select {
	case <-t.closing:
	default:
		close(t.closing)
	}
	<-t.closed
}

// Flush waits for the Tracer to flush any transactions and errors it currently
// has queued to the APM server, the tracer is stopped, or the abort channel
// is signaled.
func (t *Tracer) Flush(abort <-chan struct{}) {
	flushed := make(chan struct{}, 1)
	select {
	case t.forceFlush <- flushed:
		select {
		case <-abort:
		case <-flushed:
		case <-t.closed:
		}
	case <-t.closed:
	}
}

// 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 atomic.LoadInt32(&t.active) == 1
}

// SetRequestDuration sets the maximum amount of time to keep a request open
// to the APM server for streaming data before closing the stream and starting
// a new request.
func (t *Tracer) SetRequestDuration(d time.Duration) {
	t.sendConfigCommand(func(cfg *tracerConfig) {
		cfg.requestDuration = d
	})
}

// SetMetricsInterval sets the metrics interval -- the amount of time in
// between metrics samples being gathered.
func (t *Tracer) SetMetricsInterval(d time.Duration) {
	t.sendConfigCommand(func(cfg *tracerConfig) {
		cfg.metricsInterval = d
	})
}

// SetContextSetter sets the stacktrace.ContextSetter to be used for
// setting stacktrace source context. If nil (which is the initial
// value), no context will be set.
func (t *Tracer) SetContextSetter(setter stacktrace.ContextSetter) {
	t.sendConfigCommand(func(cfg *tracerConfig) {
		cfg.contextSetter = setter
	})
}

// SetLogger sets the Logger to be used for logging the operation of
// the tracer.
//
// If logger implements WarningLogger, its Warningf method will be used
// for logging warnings. Otherwise, warnings will logged using Debugf.
//
// The tracer is initialized with a default logger configured with the
// environment variables ELASTIC_APM_LOG_FILE and ELASTIC_APM_LOG_LEVEL.
// Calling SetLogger will replace the default logger.
func (t *Tracer) SetLogger(logger Logger) {
	t.sendConfigCommand(func(cfg *tracerConfig) {
		cfg.logger = makeWarningLogger(logger)
	})
}

// SetSanitizedFieldNames sets the wildcard patterns that will be used to
// match cookie and form field names for sanitization. Fields matching any
// of the the supplied patterns will have their values redacted. If
// SetSanitizedFieldNames is called with no arguments, then no fields
// will be redacted.
func (t *Tracer) SetSanitizedFieldNames(patterns ...string) error {
	var matchers wildcard.Matchers
	if len(patterns) != 0 {
		matchers = make(wildcard.Matchers, len(patterns))
		for i, p := range patterns {
			matchers[i] = configutil.ParseWildcardPattern(p)
		}
	}
	t.sendConfigCommand(func(cfg *tracerConfig) {
		cfg.sanitizedFieldNames = matchers
	})
	return nil
}

// RegisterMetricsGatherer registers g for periodic (or forced) metrics
// gathering by t.
//
// RegisterMetricsGatherer returns a function which will deregister g.
// It may safely be called multiple times.
func (t *Tracer) RegisterMetricsGatherer(g MetricsGatherer) func() {
	// Wrap g in a pointer-to-struct, so we can safely compare.
	wrapped := &struct{ MetricsGatherer }{MetricsGatherer: g}
	t.sendConfigCommand(func(cfg *tracerConfig) {
		cfg.metricsGatherers = append(cfg.metricsGatherers, wrapped)
	})
	deregister := func(cfg *tracerConfig) {
		for i, g := range cfg.metricsGatherers {
			if g != wrapped {
				continue
			}
			cfg.metricsGatherers = append(cfg.metricsGatherers[:i], cfg.metricsGatherers[i+1:]...)
		}
	}
	var once sync.Once
	return func() {
		once.Do(func() {
			t.sendConfigCommand(deregister)
		})
	}
}

// SetConfigWatcher sets w as the config watcher.
//
// By default, the tracer will be configured to use the transport for
// watching config, if the transport implements apmconfig.Watcher. This
// can be overridden by calling SetConfigWatcher.
//
// If w is nil, config watching will be stopped.
//
// Calling SetConfigWatcher will discard any previously observed remote
// config, reverting to local config until a config change from w is
// observed.
func (t *Tracer) SetConfigWatcher(w apmconfig.Watcher) {
	select {
	case t.configWatcher <- w:
	case <-t.closing:
	case <-t.closed:
	}
}

func (t *Tracer) sendConfigCommand(cmd tracerConfigCommand) {
	select {
	case t.configCommands <- cmd:
	case <-t.closing:
	case <-t.closed:
	}
}

// SetSampler sets the sampler the tracer.
//
// It is valid to pass nil, in which case all transactions will be sampled.
//
// Configuration via Kibana takes precedence over local configuration, so
// if sampling has been configured via Kibana, this call will not have any
// effect until/unless that configuration has been removed.
func (t *Tracer) SetSampler(s Sampler) {
	t.samplerMu.Lock()
	t.localSampler = s
	if !t.remoteSampler {
		t.sampler = s
	}
	t.samplerMu.Unlock()
}

// SetMaxSpans sets the maximum number of spans that will be added
// to a transaction before dropping spans. If set to a non-positive
// value, the number of spans is unlimited.
func (t *Tracer) SetMaxSpans(n int) {
	t.maxSpansMu.Lock()
	t.maxSpans = n
	t.maxSpansMu.Unlock()
}

// SetSpanFramesMinDuration sets the minimum duration for a span after which
// we will capture its stack frames.
func (t *Tracer) SetSpanFramesMinDuration(d time.Duration) {
	t.spanFramesMinDurationMu.Lock()
	t.spanFramesMinDuration = d
	t.spanFramesMinDurationMu.Unlock()
}

// SetStackTraceLimit sets the the maximum number of stack frames to collect
// for each stack trace. If limit is negative, then all frames will be collected.
func (t *Tracer) SetStackTraceLimit(limit int) {
	t.stackTraceLimitMu.Lock()
	t.stackTraceLimit = limit
	t.stackTraceLimitMu.Unlock()
}

// SetCaptureHeaders enables or disables capturing of HTTP headers.
func (t *Tracer) SetCaptureHeaders(capture bool) {
	t.captureHeadersMu.Lock()
	t.captureHeaders = capture
	t.captureHeadersMu.Unlock()
}

// SetCaptureBody sets the HTTP request body capture mode.
func (t *Tracer) SetCaptureBody(mode CaptureBodyMode) {
	t.captureBodyMu.Lock()
	t.captureBody = mode
	t.captureBodyMu.Unlock()
}

// SendMetrics forces the tracer to gather and send metrics immediately,
// blocking until the metrics have been sent or the abort channel is
// signalled.
func (t *Tracer) SendMetrics(abort <-chan struct{}) {
	sent := make(chan struct{}, 1)
	select {
	case t.forceSendMetrics <- sent:
		select {
		case <-abort:
		case <-sent:
		case <-t.closed:
		}
	case <-t.closed:
	}
}

// Stats returns the current TracerStats. This will return the most
// recent values even after the tracer has been closed.
func (t *Tracer) Stats() TracerStats {
	t.statsMu.Lock()
	stats := t.stats
	t.statsMu.Unlock()
	return stats
}

func (t *Tracer) loop() {
	ctx, cancelContext := context.WithCancel(context.Background())
	defer cancelContext()
	defer close(t.closed)
	defer atomic.StoreInt32(&t.active, 0)

	var req iochan.ReadRequest
	var requestBuf bytes.Buffer
	var metadata []byte
	var gracePeriod time.Duration = -1
	var flushed chan<- struct{}
	var requestBufTransactions, requestBufSpans, requestBufErrors, requestBufMetricsets uint64
	zlibWriter, _ := zlib.NewWriterLevel(&requestBuf, zlib.BestSpeed)
	zlibFlushed := true
	zlibClosed := false
	iochanReader := iochan.NewReader()
	requestBytesRead := 0
	requestActive := false
	closeRequest := false
	flushRequest := false
	requestResult := make(chan error, 1)
	requestTimer := time.NewTimer(0)
	requestTimerActive := false
	if !requestTimer.Stop() {
		<-requestTimer.C
	}

	// Run another goroutine to perform the blocking requests,
	// communicating with the tracer loop to obtain stream data.
	sendStreamRequest := make(chan time.Duration)
	defer close(sendStreamRequest)
	go func() {
		jitterRand := rand.New(rand.NewSource(time.Now().UnixNano()))
		for gracePeriod := range sendStreamRequest {
			if gracePeriod > 0 {
				select {
				case <-time.After(jitterDuration(gracePeriod, jitterRand, gracePeriodJitter)):
				case <-ctx.Done():
				}
			}
			requestResult <- t.Transport.SendStream(ctx, iochanReader)
		}
	}()

	var breakdownMetricsLimitWarningLogged bool
	var stats TracerStats
	var metrics Metrics
	var sentMetrics chan<- struct{}
	var gatheringMetrics bool
	var metricsTimerStart time.Time
	metricsBuffer := ringbuffer.New(t.metricsBufferSize)
	gatheredMetrics := make(chan struct{}, 1)
	metricsTimer := time.NewTimer(0)
	if !metricsTimer.Stop() {
		<-metricsTimer.C
	}

	var lastConfigChange map[string]string
	var configChanges <-chan apmconfig.Change
	var stopConfigWatcher func()
	defer func() {
		if stopConfigWatcher != nil {
			stopConfigWatcher()
		}
	}()

	var cfg tracerConfig
	buffer := ringbuffer.New(t.bufferSize)
	buffer.Evicted = func(h ringbuffer.BlockHeader) {
		switch h.Tag {
		case errorBlockTag:
			stats.ErrorsDropped++
		case spanBlockTag:
			stats.SpansDropped++
		case transactionBlockTag:
			stats.TransactionsDropped++
		}
	}
	modelWriter := modelWriter{
		buffer:        buffer,
		metricsBuffer: metricsBuffer,
		cfg:           &cfg,
		stats:         &stats,
	}

	for {
		var gatherMetrics bool
		select {
		case <-t.closing:
			cancelContext() // informs transport that EOF is expected
			iochanReader.CloseRead(io.EOF)
			return
		case cmd := <-t.configCommands:
			oldMetricsInterval := cfg.metricsInterval
			cmd(&cfg)
			if !gatheringMetrics && cfg.metricsInterval != oldMetricsInterval {
				if metricsTimerStart.IsZero() {
					if cfg.metricsInterval > 0 {
						metricsTimer.Reset(cfg.metricsInterval)
						metricsTimerStart = time.Now()
					}
				} else {
					if cfg.metricsInterval <= 0 {
						metricsTimerStart = time.Time{}
						if !metricsTimer.Stop() {
							<-metricsTimer.C
						}
					} else {
						alreadyPassed := time.Since(metricsTimerStart)
						if alreadyPassed >= cfg.metricsInterval {
							metricsTimer.Reset(0)
						} else {
							metricsTimer.Reset(cfg.metricsInterval - alreadyPassed)
						}
					}
				}
			}
			continue
		case cw := <-t.configWatcher:
			if configChanges != nil {
				stopConfigWatcher()
				t.updateConfig(&cfg, lastConfigChange, nil)
				lastConfigChange = nil
				configChanges = nil
			}
			if cw == nil {
				continue
			}
			var configWatcherContext context.Context
			var watchParams apmconfig.WatchParams
			watchParams.Service.Name = t.Service.Name
			watchParams.Service.Environment = t.Service.Environment
			configWatcherContext, stopConfigWatcher = context.WithCancel(ctx)
			configChanges = cw.WatchConfig(configWatcherContext, watchParams)
			// Silence go vet's "possible context leak" false positive.
			// We call a previous stopConfigWatcher before reassigning
			// the variable, and we have a defer at the top level of the
			// loop method that will call the final stopConfigWatcher
			// value on method exit.
			_ = stopConfigWatcher
			continue
		case change, ok := <-configChanges:
			if !ok {
				configChanges = nil
				continue
			}
			if change.Err != nil {
				if cfg.logger != nil {
					cfg.logger.Errorf("config request failed: %s", change.Err)
				}
			} else {
				t.updateConfig(&cfg, lastConfigChange, change.Attrs)
				lastConfigChange = change.Attrs
			}
			continue
		case event := <-t.events:
			switch event.eventType {
			case transactionEvent:
				if !t.breakdownMetrics.recordTransaction(event.tx.TransactionData) {
					if !breakdownMetricsLimitWarningLogged && cfg.logger != nil {
						cfg.logger.Warningf("%s", breakdownMetricsLimitWarning)
						breakdownMetricsLimitWarningLogged = true
					}
				}
				modelWriter.writeTransaction(event.tx.Transaction, event.tx.TransactionData)
			case spanEvent:
				modelWriter.writeSpan(event.span.Span, event.span.SpanData)
			case errorEvent:
				modelWriter.writeError(event.err)
				// Flush the buffer to transmit the error immediately.
				flushRequest = true
			}
		case <-requestTimer.C:
			requestTimerActive = false
			closeRequest = true
		case <-metricsTimer.C:
			metricsTimerStart = time.Time{}
			gatherMetrics = !gatheringMetrics
		case sentMetrics = <-t.forceSendMetrics:
			if !metricsTimerStart.IsZero() {
				if !metricsTimer.Stop() {
					<-metricsTimer.C
				}
				metricsTimerStart = time.Time{}
			}
			gatherMetrics = !gatheringMetrics
		case <-gatheredMetrics:
			modelWriter.writeMetrics(&metrics)
			gatheringMetrics = false
			flushRequest = true
			if cfg.metricsInterval > 0 {
				metricsTimerStart = time.Now()
				metricsTimer.Reset(cfg.metricsInterval)
			}
		case flushed = <-t.forceFlush:
			// Drain any objects buffered in the channels.
			for n := len(t.events); n > 0; n-- {
				event := <-t.events
				switch event.eventType {
				case transactionEvent:
					if !t.breakdownMetrics.recordTransaction(event.tx.TransactionData) {
						if !breakdownMetricsLimitWarningLogged && cfg.logger != nil {
							cfg.logger.Warningf("%s", breakdownMetricsLimitWarning)
							breakdownMetricsLimitWarningLogged = true
						}
					}
					modelWriter.writeTransaction(event.tx.Transaction, event.tx.TransactionData)
				case spanEvent:
					modelWriter.writeSpan(event.span.Span, event.span.SpanData)
				case errorEvent:
					modelWriter.writeError(event.err)
				}
			}
			if !requestActive && buffer.Len() == 0 && metricsBuffer.Len() == 0 {
				flushed <- struct{}{}
				continue
			}
			closeRequest = true
		case req = <-iochanReader.C:
		case err := <-requestResult:
			if err != nil {
				stats.Errors.SendStream++
				gracePeriod = nextGracePeriod(gracePeriod)
				if cfg.logger != nil {
					logf := cfg.logger.Debugf
					if err, ok := err.(*transport.HTTPError); ok && err.Response.StatusCode == 404 {
						// 404 typically means the server is too old, meaning
						// the error is due to a misconfigured environment.
						logf = cfg.logger.Errorf
					}
					logf("request failed: %s (next request in ~%s)", err, gracePeriod)
				}
			} else {
				gracePeriod = -1 // Reset grace period after success.
				stats.TransactionsSent += requestBufTransactions
				stats.SpansSent += requestBufSpans
				stats.ErrorsSent += requestBufErrors
				if cfg.logger != nil {
					s := func(n uint64) string {
						if n != 1 {
							return "s"
						}
						return ""
					}
					cfg.logger.Debugf(
						"sent request with %d transaction%s, %d span%s, %d error%s, %d metricset%s",
						requestBufTransactions, s(requestBufTransactions),
						requestBufSpans, s(requestBufSpans),
						requestBufErrors, s(requestBufErrors),
						requestBufMetricsets, s(requestBufMetricsets),
					)
				}
			}
			if !stats.isZero() {
				t.statsMu.Lock()
				t.stats.accumulate(stats)
				t.statsMu.Unlock()
				stats = TracerStats{}
			}
			if sentMetrics != nil && requestBufMetricsets > 0 {
				sentMetrics <- struct{}{}
				sentMetrics = nil
			}
			if flushed != nil {
				flushed <- struct{}{}
				flushed = nil
			}
			if req.Buf != nil {
				// req will be canceled by CloseRead below.
				req.Buf = nil
			}
			iochanReader.CloseRead(io.EOF)
			iochanReader = iochan.NewReader()
			flushRequest = false
			closeRequest = false
			requestActive = false
			requestBytesRead = 0
			requestBuf.Reset()
			requestBufTransactions = 0
			requestBufSpans = 0
			requestBufErrors = 0
			requestBufMetricsets = 0
			if requestTimerActive {
				if !requestTimer.Stop() {
					<-requestTimer.C
				}
				requestTimerActive = false
			}
		}

		if !stats.isZero() {
			t.statsMu.Lock()
			t.stats.accumulate(stats)
			t.statsMu.Unlock()
			stats = TracerStats{}
		}

		if gatherMetrics {
			gatheringMetrics = true
			metrics.disabled = cfg.disabledMetrics
			t.gatherMetrics(ctx, cfg.metricsGatherers, &metrics, cfg.logger, gatheredMetrics)
			if cfg.logger != nil {
				cfg.logger.Debugf("gathering metrics")
			}
		}

		if !requestActive {
			if buffer.Len() == 0 && metricsBuffer.Len() == 0 {
				continue
			}
			sendStreamRequest <- gracePeriod
			if metadata == nil {
				metadata = t.jsonRequestMetadata()
			}
			zlibWriter.Reset(&requestBuf)
			zlibWriter.Write(metadata)
			zlibFlushed = false
			zlibClosed = false
			requestActive = true
			requestTimer.Reset(cfg.requestDuration)
			requestTimerActive = true
		}

		if !closeRequest || !zlibClosed {
			for requestBytesRead+requestBuf.Len() < cfg.requestSize {
				if metricsBuffer.Len() > 0 {
					if _, _, err := metricsBuffer.WriteBlockTo(zlibWriter); err == nil {
						requestBufMetricsets++
						zlibWriter.Write([]byte("\n"))
						zlibFlushed = false
						if sentMetrics != nil {
							// SendMetrics was called: close the request
							// off so we can inform the user when the
							// metrics have been processed.
							closeRequest = true
						}
					}
					continue
				}
				if buffer.Len() == 0 {
					break
				}
				if h, _, err := buffer.WriteBlockTo(zlibWriter); err == nil {
					switch h.Tag {
					case transactionBlockTag:
						requestBufTransactions++
					case spanBlockTag:
						requestBufSpans++
					case errorBlockTag:
						requestBufErrors++
					}
					zlibWriter.Write([]byte("\n"))
					zlibFlushed = false
				}
			}
			if !closeRequest {
				closeRequest = requestBytesRead+requestBuf.Len() >= cfg.requestSize
			}
		}
		if closeRequest {
			if !zlibClosed {
				zlibWriter.Close()
				zlibClosed = true
			}
		} else if flushRequest && !zlibFlushed {
			zlibWriter.Flush()
			flushRequest = false
			zlibFlushed = true
		}

		if req.Buf == nil || requestBuf.Len() == 0 {
			continue
		}
		const zlibHeaderLen = 2
		if requestBytesRead+requestBuf.Len() > zlibHeaderLen {
			n, err := requestBuf.Read(req.Buf)
			if closeRequest && err == nil && requestBuf.Len() == 0 {
				err = io.EOF
			}
			req.Respond(n, err)
			req.Buf = nil
			if n > 0 {
				requestBytesRead += n
			}
		}
	}
}

// jsonRequestMetadata returns a JSON-encoded metadata object that features
// at the head of every request body. This is called exactly once, when the
// first request is made.
func (t *Tracer) jsonRequestMetadata() []byte {
	var json fastjson.Writer
	service := makeService(t.Service.Name, t.Service.Version, t.Service.Environment)
	json.RawString(`{"metadata":{`)
	json.RawString(`"system":`)
	t.system.MarshalFastJSON(&json)
	json.RawString(`,"process":`)
	t.process.MarshalFastJSON(&json)
	json.RawString(`,"service":`)
	service.MarshalFastJSON(&json)
	if len(globalLabels) > 0 {
		json.RawString(`,"labels":`)
		globalLabels.MarshalFastJSON(&json)
	}
	json.RawString("}}\n")
	return json.Bytes()
}

// gatherMetrics gathers metrics from each of the registered
// metrics gatherers. Once all gatherers have returned, a value
// will be sent on the "gathered" channel.
func (t *Tracer) gatherMetrics(ctx context.Context, gatherers []MetricsGatherer, m *Metrics, l Logger, gathered chan<- struct{}) {
	timestamp := model.Time(time.Now().UTC())
	var group sync.WaitGroup
	for _, g := range gatherers {
		group.Add(1)
		go func(g MetricsGatherer) {
			defer group.Done()
			gatherMetrics(ctx, g, m, l)
		}(g)
	}
	go func() {
		group.Wait()
		for _, m := range m.transactionGroupMetrics {
			m.Timestamp = timestamp
		}
		for _, m := range m.metrics {
			m.Timestamp = timestamp
		}
		gathered <- struct{}{}
	}()
}

// updateConfig updates t and cfg with changes held in "attrs", and reverts
// to local config for config attributes that have been removed (exist in old
// but not in attrs).
//
// On return from updateConfig, unapplied config will have been removed from attrs.
func (t *Tracer) updateConfig(cfg *tracerConfig, old, attrs map[string]string) {
	warningf := func(string, ...interface{}) {}
	debugf := func(string, ...interface{}) {}
	errorf := func(string, ...interface{}) {}
	if cfg.logger != nil {
		warningf = cfg.logger.Warningf
		debugf = cfg.logger.Debugf
		errorf = cfg.logger.Errorf
	}
	envName := func(k string) string {
		return "ELASTIC_APM_" + strings.ToUpper(k)
	}

	for k, v := range attrs {
		if oldv, ok := old[k]; ok && oldv == v {
			continue
		}
		switch envName(k) {
		case envTransactionSampleRate:
			sampler, err := parseSampleRate(k, v)
			if err != nil {
				errorf("central config failure: %s", err)
				delete(attrs, k)
				continue
			} else {
				t.samplerMu.Lock()
				t.sampler = sampler
				t.remoteSampler = true
				t.samplerMu.Unlock()
			}
		default:
			warningf("central config failure: unsupported config: %s", k)
			delete(attrs, k)
			continue
		}
		debugf("central config update: updated %s to %s", k, v)
	}

	for k := range old {
		if _, ok := attrs[k]; ok {
			continue
		}
		switch envName(k) {
		case envTransactionSampleRate:
			t.samplerMu.Lock()
			t.sampler = t.localSampler
			t.remoteSampler = false
			t.samplerMu.Unlock()
		default:
			continue
		}
		debugf("central config update: reverted %s to local config", k)
	}
}

type tracerEventType int

const (
	transactionEvent tracerEventType = iota
	spanEvent
	errorEvent
)

type tracerEvent struct {
	eventType tracerEventType

	// err is set only if eventType == errorEvent.
	err *ErrorData

	// tx is set only if eventType == transactionEvent.
	tx struct {
		*Transaction
		// Transaction.TransactionData is nil at the
		// point tracerEvent is created (to signify
		// that the transaction is ended), so we pass
		// it along side.
		*TransactionData
	}

	// span is set only if eventType == spanEvent.
	span struct {
		*Span
		// Span.SpanData is nil at the point tracerEvent
		// is created (to signify that the span is ended),
		// so we pass it along side.
		*SpanData
	}
}