Repository: open-telemetry/opentelemetry-go
Pull request #1757 (2021-03-30 18:56:21)

Modify ForceFlush to abort after timeout/cancellation
Modified the method, BatchSpanProcessor.ForceFlush, to abort after configured timeout or cancellation. Should resolve [Issue #1618](https://github.com/open-telemetry/opentelemetry-go/issues/1618)

linux-foundation-easycla[bot]: <a href="https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/9522512/187086161/1757/#/?version=1"><img src="https://s3.amazonaws.com/cla-project-logo-prod/cla-not-signed.svg" alt="CLA Not Signed" align="left" height="28" width="328" ></a><br/><br /><ul><li><a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/9522512/187086161/1757/#/?version=1' target='_blank'>:x:</a> - humivo The commit (a5db348672f2bb659147c486d21a04c78862186b ,98ae7727741843e3a64a7690e39455d59433d1fe ,54e1a6da436d800702a7544548a45e2dd5690f6d ,23908b4e7d1eb9d172e83d9613949f33815304ff ,65cf9343e15cc67721ee154648510b3328287a89 ,3a432e94a09d071296276067ffa2b7d764aab0b2) is not authorized under a signed CLA. <a href='https://api.easycla.lfx.linuxfoundation.org/v2/repository-provider/github/sign/9522512/187086161/1757/#/?version=1' target='_blank'>Please click here to be authorized</a>. For further assistance with EasyCLA, <a href='https://jira.linuxfoundation.org/servicedesk/customer/portal/4' target='_blank'>please submit a support request ticket</a>.</li></ul>
Aneurysm9: > --- FAIL: TestBatchSpanProcessorForceFlushSucceeds (0.00s)
    batch_span_processor_test.go:292: number of exported span: got 516, want 517

This has been a recurring issue with attempting to test this span processor.  Even though this test has set the queue size to 0 and the processor is configured to block on a full queue, it looks like it's still possible for the last span to be pulled off the queue and the scheduler to switch back to the test and start `ForceFlush` before `processQueue` finishes adding the last span to the batch.  

I wonder if testing that a non-zero number of spans were exported or that the exported count was within 10 of the expected count might be sufficient here since we simply want to know that `ForceFlush` shipped spans and the remaining spans can be picked up later.  @MrAlias or @open-telemetry/go-approvers any thoughts?

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 627d088a..7778bd88 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -41,6 +41,7 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
   The existing `ParentSpanID` and `HasRemoteParent` fields are removed in favor of this. (#1748)
 - The `ParentContext` field of the `"go.opentelemetry.io/otel/sdk/trace".SamplingParameters` is updated to hold a `context.Context` containing the parent span.
   This changes it to make `SamplingParameters` conform with the OpenTelemetry specification. (#1749)
+- Modify `BatchSpanProcessor.ForceFlush` to abort after timeout/cancellation. (#1757)
 - Improve OTLP/gRPC exporter connection errors. (#1737)
 
 ### Removed
diff --git a/sdk/trace/batch_span_processor.go b/sdk/trace/batch_span_processor.go
index aa322ae6..25d556e5 100644
--- a/sdk/trace/batch_span_processor.go
+++ b/sdk/trace/batch_span_processor.go
@@ -148,7 +148,23 @@ func (bsp *batchSpanProcessor) Shutdown(ctx context.Context) error {
 
 // ForceFlush exports all ended spans that have not yet been exported.
 func (bsp *batchSpanProcessor) ForceFlush(ctx context.Context) error {
-	return bsp.exportSpans(ctx)
+	var err error
+	if bsp.e != nil {
+		wait := make(chan struct{})
+		go func() {
+			if err := bsp.exportSpans(ctx); err != nil {
+				otel.Handle(err)
+			}
+			close(wait)
+		}()
+		// Wait until the export is finished or the context is cancelled/timed out
+		select {
+		case <-wait:
+		case <-ctx.Done():
+			err = ctx.Err()
+		}
+	}
+	return err
 }
 
 func WithMaxQueueSize(size int) BatchSpanProcessorOption {


Test patch:
diff --git a/sdk/trace/batch_span_processor_test.go b/sdk/trace/batch_span_processor_test.go
index 6b721ffe..f2bf428e 100644
--- a/sdk/trace/batch_span_processor_test.go
+++ b/sdk/trace/batch_span_processor_test.go
@@ -17,6 +17,7 @@ package trace_test
 import (
 	"context"
 	"encoding/binary"
+	"errors"
 	"sync"
 	"testing"
 	"time"
@@ -257,3 +258,71 @@ func TestBatchSpanProcessorShutdown(t *testing.T) {
 	}
 	assert.Equal(t, 1, bp.shutdownCount)
 }
+
+func TestBatchSpanProcessorForceFlushSucceeds(t *testing.T) {
+	te := testBatchExporter{}
+	tp := basicTracerProvider(t)
+	option := testOption{
+		name: "default BatchSpanProcessorOptions",
+		o: []sdktrace.BatchSpanProcessorOption{
+			sdktrace.WithMaxQueueSize(0),
+			sdktrace.WithMaxExportBatchSize(3000),
+		},
+		wantNumSpans:   2053,
+		wantBatchCount: 1,
+		genNumSpans:    2053,
+	}
+	ssp := createAndRegisterBatchSP(option, &te)
+	if ssp == nil {
+		t.Fatalf("%s: Error creating new instance of BatchSpanProcessor\n", option.name)
+	}
+	tp.RegisterSpanProcessor(ssp)
+	tr := tp.Tracer("BatchSpanProcessorWithOption")
+	generateSpan(t, option.parallel, tr, option)
+
+	// Force flush any held span batches
+	err := ssp.ForceFlush(context.Background())
+
+	gotNumOfSpans := te.len()
+	spanDifference := option.wantNumSpans - gotNumOfSpans
+	if spanDifference > 10 || spanDifference < 0 {
+		t.Errorf("number of exported span not equal to or within 10 less than: got %+v, want %+v\n",
+			gotNumOfSpans, option.wantNumSpans)
+	}
+	gotBatchCount := te.getBatchCount()
+	if gotBatchCount < option.wantBatchCount {
+		t.Errorf("number batches: got %+v, want >= %+v\n",
+			gotBatchCount, option.wantBatchCount)
+		t.Errorf("Batches %v\n", te.sizes)
+	}
+	assert.NoError(t, err)
+}
+
+func TestBatchSpanProcessorForceFlushTimeout(t *testing.T) {
+	var bp testBatchExporter
+	bsp := sdktrace.NewBatchSpanProcessor(&bp)
+	// Add timeout to context to test deadline
+	ctx, cancel := context.WithTimeout(context.Background(), time.Nanosecond)
+	defer cancel()
+	<-ctx.Done()
+
+	if err := bsp.ForceFlush(ctx); err == nil {
+		t.Error("expected context DeadlineExceeded error, got nil")
+	} else if !errors.Is(err, context.DeadlineExceeded) {
+		t.Errorf("expected context DeadlineExceeded error, got %v", err)
+	}
+}
+
+func TestBatchSpanProcessorForceFlushCancellation(t *testing.T) {
+	var bp testBatchExporter
+	bsp := sdktrace.NewBatchSpanProcessor(&bp)
+	ctx, cancel := context.WithCancel(context.Background())
+	// Cancel the context
+	cancel()
+
+	if err := bsp.ForceFlush(ctx); err == nil {
+		t.Error("expected context canceled error, got nil")
+	} else if !errors.Is(err, context.Canceled) {
+		t.Errorf("expected context canceled error, got %v", err)
+	}
+}