Poc/mpsc processor

config/config.go

@@ -32,6 +32,12 @@ type OpenTelemetry struct {
 	TLS TLS `json:"tls"`
 	// Defines the configurations to use in the sampler.
 	Sampling Sampling `json:"sampling"`
+
+	// BatchSpanProcessor configuration
+	BatchSize          int `json:"batch_size"`
+	BatchTimeout       int `json:"batch_timeout"`
+	BatchQueueSize     int `json:"batch_queue_size"`
+	BatchExportTimeout int `json:"batch_export_timeout"`
 }
 
 type TLS struct {
@@ -128,4 +134,20 @@ func (c *OpenTelemetry) SetDefaults() {
 	if c.Sampling.Type == TRACEIDRATIOBASED && c.Sampling.Rate == 0 {
 		c.Sampling.Rate = 0.5
 	}
+
+	if c.BatchSize == 0 {
+		c.BatchSize = 512
+	}
+
+	if c.BatchTimeout == 0 {
+		c.BatchTimeout = 5000
+	}
+
+	if c.BatchQueueSize == 0 {
+		c.BatchQueueSize = 2048
+	}
+
+	if c.BatchExportTimeout == 0 {
+		c.BatchExportTimeout = 30000
+	}
 }

e2e/basic/basic.go

@@ -29,6 +29,7 @@ func main() {
 		TLS: config.TLS{
 			Enable: false,
 		},
+		SpanProcessorType: "mpsc",
 	}
 
 	log.Println("Initializing OpenTelemetry at e2e-basic:", cfg.Endpoint)

trace/mpsc.go

@@ -0,0 +1,160 @@
+package trace
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"time"
+	"unsafe"
+
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+)
+
+// spanQueueNode represents a node in the queue.
+type spanQueueNode struct {
+	span sdktrace.ReadOnlySpan
+	next *spanQueueNode
+}
+
+// spanQueue is a basic lock-free queue for spans.
+type spanQueue struct {
+	head    atomic.Pointer[spanQueueNode]
+	tail    atomic.Pointer[spanQueueNode]
+	padding [128]byte // Padding to avoid false sharing between head and tail
+}
+
+// newSpanQueue creates a new spanQueue.
+func newSpanQueue() *spanQueue {
+	q := &spanQueue{}
+	node := &spanQueueNode{} // Dummy node
+	q.head.Store(node)
+	q.tail.Store(node)
+	return q
+}
+
+// enqueue adds a span to the queue.
+func (q *spanQueue) enqueue(span sdktrace.ReadOnlySpan) {
+	node := &spanQueueNode{span: span}
+	for {
+		tail := q.tail.Load()
+		tailNext := (*unsafe.Pointer)(unsafe.Pointer(tail.next))
+		if atomic.CompareAndSwapPointer(tailNext, nil, unsafe.Pointer(node)) {
+			q.tail.CompareAndSwap(tail, node)
+			return
+		}
+		q.tail.CompareAndSwap(tail, tail.next)
+	}
+}
+
+// dequeue removes and returns the next span from the queue.
+func (q *spanQueue) dequeue() (sdktrace.ReadOnlySpan, bool) {
+	for {
+		head := q.head.Load()
+		next := head.next
+		if next == nil {
+			return nil, false // Queue is empty
+		}
+		if q.head.CompareAndSwap(head, next) {
+			return next.span, true
+		}
+	}
+}
+
+// BatchSpanProcessor is an implementation of the SpanProcessor that batches spans for async processing.
+type BatchSpanProcessor struct {
+	queue      *spanQueue
+	maxBatch   int
+	exporter   sdktrace.SpanExporter
+	shutdownCh chan struct{}
+	wg         sync.WaitGroup
+}
+
+// NewBatchSpanProcessor creates a new BatchSpanProcessor.
+func NewMPSCSpanProcessor(exporter sdktrace.SpanExporter, maxBatchSize int) *BatchSpanProcessor {
+	bsp := &BatchSpanProcessor{
+		queue:      newSpanQueue(),
+		maxBatch:   maxBatchSize,
+		exporter:   exporter,
+		shutdownCh: make(chan struct{}),
+	}
+
+	bsp.wg.Add(1)
+	go bsp.processQueue()
+
+	return bsp
+}
+
+// OnStart is called when a span is started.
+func (bsp *BatchSpanProcessor) OnStart(_ context.Context, _ sdktrace.ReadWriteSpan) {
+	// Do nothing on start.
+}
+
+// OnEnd is called when a span is finished.
+func (bsp *BatchSpanProcessor) OnEnd(s sdktrace.ReadOnlySpan) {
+	bsp.queue.enqueue(s)
+}
+
+// Shutdown is called when the SDK shuts down.
+func (bsp *BatchSpanProcessor) Shutdown(ctx context.Context) error {
+	close(bsp.shutdownCh)
+	bsp.wg.Wait()
+
+	return bsp.exporter.Shutdown(ctx)
+}
+
+// ForceFlush exports all ended spans that have not yet been exported.
+func (bsp *BatchSpanProcessor) ForceFlush(ctx context.Context) error {
+	batch := bsp.collectBatch()
+	if len(batch) > 0 {
+		return bsp.exporter.ExportSpans(ctx, batch)
+	}
+	return nil
+}
+
+// processQueue processes the span queue in batches.
+func (bsp *BatchSpanProcessor) processQueue() {
+	defer bsp.wg.Done()
+
+	ticker := time.NewTicker(200 * time.Millisecond)
+	defer ticker.Stop()
+
+	batch := make([]sdktrace.ReadOnlySpan, 0, bsp.maxBatch)
+
+	for {
+		select {
+		case <-bsp.shutdownCh:
+			return
+		case <-ticker.C:
+			if len(batch) > 0 {
+				bsp.exportBatch(batch)
+				batch = make([]sdktrace.ReadOnlySpan, 0, bsp.maxBatch)
+			}
+		default:
+			if span, ok := bsp.queue.dequeue(); ok {
+				batch = append(batch, span)
+				if len(batch) >= bsp.maxBatch {
+					bsp.exportBatch(batch)
+					batch = make([]sdktrace.ReadOnlySpan, 0, bsp.maxBatch)
+				}
+			}
+		}
+	}
+}
+
+// collectBatch collects a batch of spans from the queue.
+func (bsp *BatchSpanProcessor) collectBatch() []sdktrace.ReadOnlySpan {
+	var batch []sdktrace.ReadOnlySpan
+	for {
+		if span, ok := bsp.queue.dequeue(); ok {
+			batch = append(batch, span)
+		} else {
+			break
+		}
+	}
+	return batch
+}
+
+// exportBatch exports a batch of spans.
+func (bsp *BatchSpanProcessor) exportBatch(batch []sdktrace.ReadOnlySpan) {
+	_ = bsp.exporter.ExportSpans(context.Background(), batch)
+}

trace/provider.go

@@ -99,7 +99,7 @@ func NewProvider(opts ...Option) (Provider, error) {
 	}
 
 	// create the span processor - this is what will send the spans to the exporter.
-	spanProcesor := spanProcessorFactory(provider.cfg.SpanProcessorType, exporter)
+	spanProcesor := spanProcessorFactory(provider.cfg.SpanProcessorType, exporter, provider.cfg)
 
 	// create the sampler based on the configs
 	samplerType := provider.cfg.Sampling.Type

trace/span_processor.go

@@ -1,23 +1,44 @@
 package trace
 
 import (
+	"fmt"
+	"time"
+
+	"github.com/TykTechnologies/opentelemetry/config"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 )
 
-func spanProcessorFactory(spanProcessorType string, exporter sdktrace.SpanExporter) sdktrace.SpanProcessor {
+func spanProcessorFactory(spanProcessorType string, exporter sdktrace.SpanExporter, cfg *config.OpenTelemetry) sdktrace.SpanProcessor {
 	switch spanProcessorType {
 	case "simple":
 		return newSimpleSpanProcessor(exporter)
+	case "mpsc":
+		fmt.Println("Using MPSC Span Processor")
+		return NewMPSCSpanProcessor(exporter, cfg.BatchSize)
 	default:
 		// Default to BatchSpanProcessor
-		return newBatchSpanProcessor(exporter)
+		return newBatchSpanProcessor(exporter, cfg)
 	}
 }
 
 func newSimpleSpanProcessor(exporter sdktrace.SpanExporter) sdktrace.SpanProcessor {
 	return sdktrace.NewSimpleSpanProcessor(exporter)
 }
 
-func newBatchSpanProcessor(exporter sdktrace.SpanExporter) sdktrace.SpanProcessor {
-	return sdktrace.NewBatchSpanProcessor(exporter)
+func newBatchSpanProcessor(exporter sdktrace.SpanExporter, cfg *config.OpenTelemetry) sdktrace.SpanProcessor {
+	opts := []sdktrace.BatchSpanProcessorOption{}
+	if cfg.BatchSize > 0 {
+		opts = append(opts, sdktrace.WithMaxExportBatchSize(cfg.BatchSize))
+	}
+	if cfg.BatchQueueSize > 0 {
+		opts = append(opts, sdktrace.WithMaxQueueSize(cfg.BatchQueueSize))
+	}
+	if cfg.BatchTimeout > 0 {
+		opts = append(opts, sdktrace.WithBatchTimeout(time.Duration(cfg.BatchTimeout)*time.Millisecond))
+	}
+	if cfg.BatchExportTimeout > 0 {
+		opts = append(opts, sdktrace.WithExportTimeout(time.Duration(cfg.BatchExportTimeout)*time.Millisecond))
+	}
+
+	return sdktrace.NewBatchSpanProcessor(exporter, opts...)
 }

trace/span_processor_test.go

@@ -5,6 +5,7 @@ import (
 	"strconv"
 	"testing"
 
+	"github.com/TykTechnologies/opentelemetry/config"
 	"github.com/stretchr/testify/assert"
 	sdktrace "go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/trace"
@@ -76,7 +77,7 @@ func Test_NewBatchSpanProcessor(t *testing.T) {
 		te := testExporter{}
 
 		// Create a new span processor
-		processor := newBatchSpanProcessor(&te)
+		processor := newBatchSpanProcessor(&te, &config.OpenTelemetry{})
 		assert.NotNil(t, processor)
 		// Create a new tracer provider
 		tp := sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.AlwaysSample()))
@@ -104,7 +105,7 @@ func Test_NewBatchSpanProcessor(t *testing.T) {
 		te := testExporter{}
 
 		// Create a new span processor
-		processor := newBatchSpanProcessor(&te)
+		processor := newBatchSpanProcessor(&te, &config.OpenTelemetry{})
 		assert.NotNil(t, processor)
 		// Create a new tracer provider
 		tp := sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.AlwaysSample()))
@@ -142,7 +143,7 @@ func Test_NewBatchSpanProcessor(t *testing.T) {
 		te := testExporter{}
 
 		// Create a new span processor
-		processor := newBatchSpanProcessor(&te)
+		processor := newBatchSpanProcessor(&te, &config.OpenTelemetry{})
 		assert.NotNil(t, processor)
 		// Create a new tracer provider
 		tp := sdktrace.NewTracerProvider(sdktrace.WithSampler(sdktrace.AlwaysSample()))