From 3c7facee73444f8deef39ba20afb99e13e2e2e75 Mon Sep 17 00:00:00 2001 From: Matej Gera <38492574+matej-g@users.noreply.github.com> Date: Mon, 5 Apr 2021 19:38:03 +0200 Subject: [PATCH] Add `ExportTimeout` option to batch span processor (#1755) * Add ExportTimeout option * Adjust tests * Update CHANGELOG * Beef up the exporter timeout test * Beef up exporter test - attempt #2 Co-authored-by: Tyler Yahn --- CHANGELOG.md | 1 + sdk/trace/batch_span_processor.go | 19 ++++++++++ sdk/trace/batch_span_processor_test.go | 48 +++++++++++++++++++++----- 3 files changed, 59 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a059c08cd1f..7f4a44b60aa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ### Added +- Option `ExportTimeout` was added to batch span processor. (#1755) - Adds semantic conventions for exceptions. (#1492) - Added support for configuring OTLP/HTTP Endpoints, Headers, Compression and Timeout via the Environment Variables. (#1758) - `OTEL_EXPORTER_OTLP_ENDPOINT` diff --git a/sdk/trace/batch_span_processor.go b/sdk/trace/batch_span_processor.go index 25d556e5f35..a2494135f4d 100644 --- a/sdk/trace/batch_span_processor.go +++ b/sdk/trace/batch_span_processor.go @@ -28,6 +28,7 @@ import ( const ( DefaultMaxQueueSize = 2048 DefaultBatchTimeout = 5000 * time.Millisecond + DefaultExportTimeout = 30000 * time.Millisecond DefaultMaxExportBatchSize = 512 ) @@ -44,6 +45,11 @@ type BatchSpanProcessorOptions struct { // The default value of BatchTimeout is 5000 msec. BatchTimeout time.Duration + // ExportTimeout specifies the maximum duration for exporting spans. If the timeout + // is reached, the export will be cancelled. + // The default value of ExportTimeout is 30000 msec. + ExportTimeout time.Duration + // MaxExportBatchSize is the maximum number of spans to process in a single batch. // If there are more than one batch worth of spans then it processes multiple batches // of spans one batch after the other without any delay. @@ -83,6 +89,7 @@ var _ SpanProcessor = (*batchSpanProcessor)(nil) func NewBatchSpanProcessor(exporter export.SpanExporter, options ...BatchSpanProcessorOption) SpanProcessor { o := BatchSpanProcessorOptions{ BatchTimeout: DefaultBatchTimeout, + ExportTimeout: DefaultExportTimeout, MaxQueueSize: DefaultMaxQueueSize, MaxExportBatchSize: DefaultMaxExportBatchSize, } @@ -185,6 +192,12 @@ func WithBatchTimeout(delay time.Duration) BatchSpanProcessorOption { } } +func WithExportTimeout(timeout time.Duration) BatchSpanProcessorOption { + return func(o *BatchSpanProcessorOptions) { + o.ExportTimeout = timeout + } +} + func WithBlocking() BatchSpanProcessorOption { return func(o *BatchSpanProcessorOptions) { o.BlockOnQueueFull = true @@ -198,6 +211,12 @@ func (bsp *batchSpanProcessor) exportSpans(ctx context.Context) error { bsp.batchMutex.Lock() defer bsp.batchMutex.Unlock() + if bsp.o.ExportTimeout > 0 { + var cancel context.CancelFunc + ctx, cancel = context.WithTimeout(ctx, bsp.o.ExportTimeout) + defer cancel() + } + if len(bsp.batch) > 0 { if err := bsp.e.ExportSpans(ctx, bsp.batch); err != nil { return err diff --git a/sdk/trace/batch_span_processor_test.go b/sdk/trace/batch_span_processor_test.go index f2bf428e22a..44c3c2f69f2 100644 --- a/sdk/trace/batch_span_processor_test.go +++ b/sdk/trace/batch_span_processor_test.go @@ -36,12 +36,23 @@ type testBatchExporter struct { sizes []int batchCount int shutdownCount int + delay time.Duration + err error } func (t *testBatchExporter) ExportSpans(ctx context.Context, ss []*export.SpanSnapshot) error { t.mu.Lock() defer t.mu.Unlock() + time.Sleep(t.delay) + + select { + case <-ctx.Done(): + t.err = ctx.Err() + return ctx.Err() + default: + } + t.spans = append(t.spans, ss...) t.sizes = append(t.sizes, len(ss)) t.batchCount++ @@ -88,16 +99,19 @@ func TestNewBatchSpanProcessorWithNilExporter(t *testing.T) { } type testOption struct { - name string - o []sdktrace.BatchSpanProcessorOption - wantNumSpans int - wantBatchCount int - genNumSpans int - parallel bool + name string + o []sdktrace.BatchSpanProcessorOption + wantNumSpans int + wantBatchCount int + wantExportTimeout bool + genNumSpans int + delayExportBy time.Duration + parallel bool } func TestNewBatchSpanProcessorWithOptions(t *testing.T) { schDelay := 200 * time.Millisecond + exportTimeout := time.Millisecond options := []testOption{ { name: "default BatchSpanProcessorOptions", @@ -105,6 +119,15 @@ func TestNewBatchSpanProcessorWithOptions(t *testing.T) { wantBatchCount: 4, genNumSpans: 2053, }, + { + name: "non-default ExportTimeout", + o: []sdktrace.BatchSpanProcessorOption{ + sdktrace.WithExportTimeout(exportTimeout), + }, + wantExportTimeout: true, + genNumSpans: 2053, + delayExportBy: 2 * exportTimeout, // to ensure export timeout + }, { name: "non-default BatchTimeout", o: []sdktrace.BatchSpanProcessorOption{ @@ -171,7 +194,9 @@ func TestNewBatchSpanProcessorWithOptions(t *testing.T) { } for _, option := range options { t.Run(option.name, func(t *testing.T) { - te := testBatchExporter{} + te := testBatchExporter{ + delay: option.delayExportBy, + } tp := basicTracerProvider(t) ssp := createAndRegisterBatchSP(option, &te) if ssp == nil { @@ -185,17 +210,22 @@ func TestNewBatchSpanProcessorWithOptions(t *testing.T) { tp.UnregisterSpanProcessor(ssp) gotNumOfSpans := te.len() - if option.wantNumSpans != gotNumOfSpans { + if option.wantNumSpans > 0 && option.wantNumSpans != gotNumOfSpans { t.Errorf("number of exported span: got %+v, want %+v\n", gotNumOfSpans, option.wantNumSpans) } gotBatchCount := te.getBatchCount() - if gotBatchCount < option.wantBatchCount { + if option.wantBatchCount > 0 && gotBatchCount < option.wantBatchCount { t.Errorf("number batches: got %+v, want >= %+v\n", gotBatchCount, option.wantBatchCount) t.Errorf("Batches %v\n", te.sizes) } + + if option.wantExportTimeout && te.err != context.DeadlineExceeded { + t.Errorf("context deadline: got err %+v, want %+v\n", + te.err, context.DeadlineExceeded) + } }) } }