TT-10413 Poc/processors

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
+	}
 }

config/config_test.go

@@ -35,6 +35,10 @@ func Test_SetDefault(t *testing.T) {
 					Type: TRACEIDRATIOBASED,
 					Rate: 0.8,
 				},
+				BatchSize:          1,
+				BatchTimeout:       1,
+				BatchQueueSize:     2,
+				BatchExportTimeout: 2,
 			},
 			expectedCfg: OpenTelemetry{
 				Enabled:            true,
@@ -48,6 +52,10 @@ func Test_SetDefault(t *testing.T) {
 					Type: TRACEIDRATIOBASED,
 					Rate: 0.8,
 				},
+				BatchSize:          1,
+				BatchTimeout:       1,
+				BatchQueueSize:     2,
+				BatchExportTimeout: 2,
 			},
 		},
 		{
@@ -66,6 +74,10 @@ func Test_SetDefault(t *testing.T) {
 				Sampling: Sampling{
 					Type: ALWAYSON,
 				},
+				BatchSize:          512,
+				BatchTimeout:       5000,
+				BatchQueueSize:     2048,
+				BatchExportTimeout: 30000,
 			},
 		},
 		{
@@ -88,6 +100,10 @@ func Test_SetDefault(t *testing.T) {
 					Type: TRACEIDRATIOBASED,
 					Rate: 0.5,
 				},
+				BatchSize:          512,
+				BatchTimeout:       5000,
+				BatchQueueSize:     2048,
+				BatchExportTimeout: 30000,
 			},
 		},
 	}

e2e/basic/basic.go

@@ -3,10 +3,12 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"log"
 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"
 
@@ -16,6 +18,31 @@ import (
 	"github.com/sirupsen/logrus"
 )
 
+// processEnvs is a helper function to process environment variables - usefull for development.
+// e.g ENDPOINT="localhost:4317" PROCESSOR="mpsc" BATCH_SIZE=1 go run basic.go
+func processEnvs(cfg *config.OpenTelemetry) {
+	if processor := os.Getenv("PROCESSOR"); processor != "" {
+		cfg.SpanProcessorType = processor
+		fmt.Println("using custom processor:", processor)
+	}
+
+	if batchSize := os.Getenv("BATCH_SIZE"); batchSize != "" {
+		cfg.BatchSize, _ = strconv.Atoi(batchSize)
+
+		fmt.Println("using custom batch size:", batchSize)
+	}
+
+	if batchTimeout := os.Getenv("BATCH_TIMEOUT"); batchTimeout != "" {
+		cfg.BatchTimeout, _ = strconv.Atoi(batchTimeout)
+		fmt.Println("using custom batch timeout:", batchTimeout)
+	}
+
+	if endpoint := os.Getenv("ENDPOINT"); endpoint != "" {
+		cfg.Endpoint = endpoint
+		fmt.Println("using custom endpoint:", endpoint)
+	}
+}
+
 func main() {
 	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM, syscall.SIGINT)
 	defer cancel()
@@ -31,6 +58,8 @@ func main() {
 		},
 	}
 
+	processEnvs(&cfg)
+
 	log.Println("Initializing OpenTelemetry at e2e-basic:", cfg.Endpoint)
 
 	provider, err := trace.NewProvider(

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,45 @@
 package trace
 
 import (
+	"time"
+
+	"github.com/TykTechnologies/opentelemetry/config"
+	"github.com/TykTechnologies/opentelemetry/trace/sprocessor"
 	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 "tyk":
+		return sprocessor.NewAnalyticsHandler(exporter, cfg)
+	case "mpsc":
+		return sprocessor.NewMPSCSpanProcessor(exporter, cfg.BatchSize, cfg.BatchTimeout)
 	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()))