batch_span_processor_v2.go

// Copyright The OpenTelemetry Authors
//
// Licensed 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 trace // import "go.opentelemetry.io/otel/sdk/trace"

import (
	"context"
	"time"
)

const (
	DefaultV2MaxQueueSize       = 2048
	DefaultV2BatchTimeout       = 5000 * time.Millisecond
	DefaultV2ExportTimeout      = 30000 * time.Millisecond
	DefaultV2MaxExportBatchSize = 512
)

type batchSpanProcessorV2 struct {
	e SpanExporter
	o BatchSpanProcessorOptions

	queue   chan bspV2State
	stopped bool

	batch []ReadOnlySpan
	timer *time.Timer
}

type bspV2StateKind int

const (
	bspV2Msg = iota
	bspV2Flush
	bspV2Close
)

type bspV2State struct {
	kind bspV2StateKind
	span ReadOnlySpan
	ctx  context.Context
	sync chan error
}

var _ SpanProcessor = (*batchSpanProcessorV2)(nil)

func NewBatchSpanProcessorV2(exporter SpanExporter, options ...BatchSpanProcessorOption) SpanProcessor {
	o := BatchSpanProcessorOptions{
		BatchTimeout:       DefaultV2BatchTimeout,
		ExportTimeout:      DefaultV2ExportTimeout,
		MaxQueueSize:       DefaultV2MaxQueueSize,
		MaxExportBatchSize: DefaultV2MaxExportBatchSize,
	}
	for _, opt := range options {
		opt(&o)
	}

	bsp := &batchSpanProcessorV2{
		e:     exporter,
		o:     o,
		batch: make([]ReadOnlySpan, 0, o.MaxExportBatchSize),
		timer: time.NewTimer(o.BatchTimeout),
		queue: make(chan bspV2State, o.MaxQueueSize),
	}

	go bsp.processQueue()

	return bsp
}

// OnStart is called when a span is started. It is called synchronously
// and should not block.
func (bsp *batchSpanProcessorV2) OnStart(parent context.Context, s ReadWriteSpan) {}

// OnEnd is called when span is finished. It is called synchronously and
// hence not block.
func (bsp *batchSpanProcessorV2) OnEnd(s ReadOnlySpan) {
	if bsp.e == nil {
		return
	}
	state := bspV2State{
		kind: bspV2Msg,
		span: s,
	}

	if bsp.o.BlockOnQueueFull {
		bsp.queue <- state
		return
	}

	select {
	case bsp.queue <- state:
	default:
	}
}

// Shutdown is called when the SDK shuts down. Any cleanup or release of
// resources held by the processor should be done in this call.
//
// Calls to OnStart, OnEnd, or ForceFlush after this has been called
// should be ignored.
//
// All timeouts and cancellations contained in ctx must be honored, this
// should not block indefinitely.
func (bsp *batchSpanProcessorV2) Shutdown(ctx context.Context) error {
	errCh := make(chan error)
	bsp.queue <- bspV2State{
		kind: bspV2Close,
		sync: errCh,
		ctx:  ctx,
	}
	return <-errCh
}

func (bsp *batchSpanProcessorV2) AsyncShutdown(ctx context.Context) error {
	bsp.queue <- bspV2State{
		kind: bspV2Close,
		ctx:  ctx,
	}
	return nil
}

// ForceFlush exports all ended spans to the configured Exporter that have not yet
// been exported.  It should only be called when absolutely necessary, such as when
// using a FaaS provider that may suspend the process after an invocation, but before
// the Processor can export the completed spans.
func (bsp *batchSpanProcessorV2) ForceFlush(ctx context.Context) error {
	errCh := make(chan error)
	bsp.queue <- bspV2State{
		kind: bspV2Flush,
		sync: errCh,
		ctx:  ctx,
	}
	return <-errCh
}

func (bsp *batchSpanProcessorV2) AsyncForceFlush(ctx context.Context) error {
	errCh := make(chan error)
	bsp.queue <- bspV2State{
		kind: bspV2Flush,
		sync: errCh,
		ctx:  ctx,
	}
	return <-errCh
}

func (bsp *batchSpanProcessorV2) processQueue() {
	for {
		select {
		case <-bsp.timer.C:
			bsp.exportSpans(context.Background())
		case msg := <-bsp.queue:

			switch msg.kind {
			case bspV2Msg:
				if bsp.stopped {
					continue
				}

				bsp.batch = append(bsp.batch, msg.span)

				if len(bsp.batch) >= bsp.o.MaxExportBatchSize {
					bsp.exportSpans(context.Background())
				}

			case bspV2Flush:
				err := bsp.exportSpans(msg.ctx)
				if msg.sync != nil {
					msg.sync <- err
					close(msg.sync)
				}

			case bspV2Close:
				err := bsp.exportSpans(msg.ctx)
				if msg.sync != nil {
					msg.sync <- err
					close(msg.sync)
				}
				if bsp.stopped {
					continue
				}
				bsp.timer.Stop()
				bsp.stopped = true
				if bsp.e != nil {
					bsp.e.Shutdown(msg.ctx)
				}
			}
		}

	}
}

func (bsp *batchSpanProcessorV2) exportSpans(ctx context.Context) (err error) {
	if bsp.stopped {
		return nil
	}
	if l := len(bsp.batch); l > 0 {
		err = bsp.e.ExportSpans(ctx, bsp.batch)
		//TODO: handle error

		// A new batch is always created after exporting, even if the batch failed to be exported.
		//
		// It is up to the exporter to implement any type of retry logic if a batch is failing
		// to be exported, since it is specific to the protocol and backend being sent to.
		bsp.batch = bsp.batch[:0]

	}

	if !bsp.timer.Stop() {
		<-bsp.timer.C
	}
	bsp.timer.Reset(bsp.o.BatchTimeout)

	return err
}