Repository: open-telemetry/opentelemetry-go
Pull request #766 (2020-05-26 07:19:10)

Fix BatchSpanProcessor.Shutdown to wait until all spans are processed
Currently it exits too soon - before drainQueue is finished

Aneurysm9: Is there a test case that can be added to demonstrate the faulty behavior and ensure it isn't reintroduced?
jmacd: This might never exit, I think. 

I believe it's WAI. If you shutdown the processor while there are still spans being generated, some of them will not be recorded. What are the semantics you want, here?
vmihailenco: I would like to be able to understand what is "working as intented" here and how it might never exit. That is not first time I hear that from you so I guess I should point out that it does not help in any way except politely says "no" without any explanation (keep guessing).

>If you shutdown the processor while there are still spans being generated, some of them will not be recorded

Given that after shutdown spans are just not added to the internal queue (channel) I don't see what to discuss here. You are right - they are discarded.

>What are the semantics you want, here?

I am not particularly interested in any behavior here. But the code does not do what the comment says and what common sense suggests. Perhaps it makes sense to update the comment. Closing since it is WAI and I don't use this.
jmacd: I could be wrong, but it looked like drainQueue() would continue to process spans as long sufficiently many of them are generated to never fully drain the queue. I think we should expose a `Flush()` operation to flush the processor -- there is also a spec issue written on this topic -- that would allow a user to flush and then shutdown without losing spans.
vmihailenco: That is supposed to be handled by stopCh in enqueue - otherwise it is not clear why to have it there in first place.
jmacd: You're correct. It's clear that I'm not providing an effective review here, and I'm sorry. I've explained why I would not use this code (because it does not provide a way to limit encoded batch size), and that I only became involved because flaky tests were interfering with the metrics SDK development. 

It worries me that we have disabled the tests for this class and are still making changes. What you point out is true, and we could either remove `stopCh` and `drainQueue` or we could make this change. I am concerned that even with this change, the existing test wouldn't pass. 

The underlying problem seems to be that there is no "correct" way to shut down a processor when there are concurrent writers. I now believe that adding Flush() is the best path forward. We can write a correctness test for Flush() that ensures all spans are written, then all we need to test is that Shutdown closes channels, doesn't block indefinitely, and doesn't panic.
vmihailenco: I see at least 2 reasons why tests were flaky:
- Shutdown exits before drainQueue is finished.
- some tests don't use `WithBlocking` so spans can be dropped if scheduling is lucky/unlucky

This PR fixes both so hopefully tests should pass again.

>The underlying problem seems to be that there is no "correct" way to shut down a processor when there are concurrent writers. I now believe that adding Flush() is the best path forward. 

`Shutdown` only guarantees that spans added before the shutdown are processed. It does not help with concurrent writers - you should stop/synchronize them separately.

Tests should work fine because all spans are added before calling `Shutdown` and now they use `WithBlocking` so spans can't be dropped when queue is full.
jmacd: The last time I wrangled with this topic, I did add `WithBlocking()` in the tests, but I wasn't able to make the test pass. We discussed this in the SIG call today, and I think the best we could do in terms of giving the user proper semantics, would be to add a Flush() call (with tests) and then simplify Shutdown.
communitybridge-easycla[bot]: [![CLA Check](https://s3.amazonaws.com/cla-project-logo-prod/cla-notsigned.png)](https://api.lfcla.com/v2/repository-provider/github/sign/9522512/187086161/766)<br /><ul><li>:white_check_mark:  Joshua MacDonald (00dd83dd809a7fb1089edb46555ada4a061615e8)</li></ul><ul><li>[:x:](https://api.lfcla.com/v2/repository-provider/github/sign/9522512/187086161/766)  Vladimir Mihailenco The commit (44cfb0699133a5c624b2cad84dec7b6175d59257 ,5a80a0b8612a7328026263f04cef2cadf2f4121b ,63d524e17a707d6a4fc8cab2d14e7160d595a333 ,09c277d76d6af09a23fd69b3dd8eccf107d415a3) is not authorized under a signed CLA. [Please click here to be authorized](https://api.lfcla.com/v2/repository-provider/github/sign/9522512/187086161/766). For further assistance with EasyCLA, [please submit a support request ticket](https://jira.linuxfoundation.org/servicedesk/customer/portal/4).</li></ul>
jmacd: Thank you @vmihailenco, sorry for the delay, and sorry for several mistakes of mine in reviewing this code. 
jmacd: @lizthegrey as our compliance officer, would you look at the error above (https://github.com/open-telemetry/opentelemetry-go/pull/766#issuecomment-640930634)? I can't see why the build is green given that message.
jmacd: @lizthegrey nevermind. I see you posted on another PR that we have to re-authorize the CLA.
However, this makes me think the test is broken because the build is green. Thoughts?
jmacd: This restores tests! Merging.

Patch:
diff --git a/sdk/trace/batch_span_processor.go b/sdk/trace/batch_span_processor.go
index 4030e0a3..db647d97 100644
--- a/sdk/trace/batch_span_processor.go
+++ b/sdk/trace/batch_span_processor.go
@@ -105,9 +105,10 @@ func NewBatchSpanProcessor(e export.SpanBatcher, opts ...BatchSpanProcessorOptio
 		queue:  make(chan *export.SpanData, o.MaxQueueSize),
 		stopCh: make(chan struct{}),
 	}
-	bsp.stopWait.Add(1)
 
+	bsp.stopWait.Add(1)
 	go func() {
+		defer bsp.stopWait.Done()
 		bsp.processQueue()
 		bsp.drainQueue()
 	}()
@@ -130,8 +131,6 @@ func (bsp *BatchSpanProcessor) Shutdown() {
 	bsp.stopOnce.Do(func() {
 		close(bsp.stopCh)
 		bsp.stopWait.Wait()
-		close(bsp.queue)
-
 	})
 }
 
@@ -173,7 +172,6 @@ func (bsp *BatchSpanProcessor) exportSpans() {
 // is shut down. It calls the exporter in batches of up to MaxExportBatchSize
 // waiting up to BatchTimeout to form a batch.
 func (bsp *BatchSpanProcessor) processQueue() {
-	defer bsp.stopWait.Done()
 	defer bsp.timer.Stop()
 
 	for {
@@ -197,13 +195,22 @@ func (bsp *BatchSpanProcessor) processQueue() {
 // drainQueue awaits the any caller that had added to bsp.stopWait
 // to finish the enqueue, then exports the final batch.
 func (bsp *BatchSpanProcessor) drainQueue() {
-	for sd := range bsp.queue {
-		bsp.batch = append(bsp.batch, sd)
-		if len(bsp.batch) == bsp.o.MaxExportBatchSize {
-			bsp.exportSpans()
+	for {
+		select {
+		case sd := <-bsp.queue:
+			if sd == nil {
+				bsp.exportSpans()
+				return
+			}
+
+			bsp.batch = append(bsp.batch, sd)
+			if len(bsp.batch) == bsp.o.MaxExportBatchSize {
+				bsp.exportSpans()
+			}
+		default:
+			close(bsp.queue)
 		}
 	}
-	bsp.exportSpans()
 }
 
 func (bsp *BatchSpanProcessor) enqueue(sd *export.SpanData) {
@@ -226,17 +233,19 @@ func (bsp *BatchSpanProcessor) enqueue(sd *export.SpanData) {
 		panic(x)
 	}()
 
+	select {
+	case <-bsp.stopCh:
+		return
+	default:
+	}
+
 	if bsp.o.BlockOnQueueFull {
-		select {
-		case bsp.queue <- sd:
-		case <-bsp.stopCh:
-		}
+		bsp.queue <- sd
 		return
 	}
 
 	select {
 	case bsp.queue <- sd:
-	case <-bsp.stopCh:
 	default:
 		atomic.AddUint32(&bsp.dropped, 1)
 	}


Test patch:
diff --git a/sdk/trace/batch_span_processor_test.go b/sdk/trace/batch_span_processor_test.go
index 3e80a20d..373cc18b 100644
--- a/sdk/trace/batch_span_processor_test.go
+++ b/sdk/trace/batch_span_processor_test.go
@@ -117,7 +117,6 @@ func TestNewBatchSpanProcessorWithOptions(t *testing.T) {
 				sdktrace.WithBatchTimeout(schDelay),
 				sdktrace.WithMaxQueueSize(200),
 				sdktrace.WithMaxExportBatchSize(20),
-				sdktrace.WithBlocking(),
 			},
 			wantNumSpans:   205,
 			wantBatchCount: 11,
@@ -139,7 +138,6 @@ func TestNewBatchSpanProcessorWithOptions(t *testing.T) {
 			o: []sdktrace.BatchSpanProcessorOption{
 				sdktrace.WithBatchTimeout(schDelay),
 				sdktrace.WithMaxExportBatchSize(200),
-				sdktrace.WithBlocking(),
 			},
 			wantNumSpans:   2000,
 			wantBatchCount: 10,
@@ -162,18 +160,26 @@ func TestNewBatchSpanProcessorWithOptions(t *testing.T) {
 
 			tp.UnregisterSpanProcessor(ssp)
 
-			// TODO(https://github.com/open-telemetry/opentelemetry-go/issues/741)
-			// Restore some sort of test here.
-			_ = option.wantNumSpans
-			_ = option.wantBatchCount
-			_ = te.len()           // gotNumOfSpans
-			_ = te.getBatchCount() // gotBatchCount
+			gotNumOfSpans := te.len()
+			if option.wantNumSpans != gotNumOfSpans {
+				t.Errorf("number of exported span: 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)
+			}
 		})
 	}
 }
 
 func createAndRegisterBatchSP(t *testing.T, option testOption, te *testBatchExporter) *sdktrace.BatchSpanProcessor {
-	ssp, err := sdktrace.NewBatchSpanProcessor(te, option.o...)
+	// Always use blocking queue to avoid flaky tests.
+	options := append(option.o, sdktrace.WithBlocking())
+	ssp, err := sdktrace.NewBatchSpanProcessor(te, options...)
 	if ssp == nil {
 		t.Errorf("%s: Error creating new instance of BatchSpanProcessor, error: %v\n", option.name, err)
 	}