sdk: Implement force_flush for span processors (#389)

open-telemetry/opentelemetry-specification#370 added
the requirement to have a "force_flush" method in the span processors.

This commit exposes an already existing internal method on the batch span
processor that does exactly the same, it also adds it to the span processor
interface and as a no-op to the simple span processor.

opentelemetry-sdk/src/opentelemetry/sdk/trace/__init__.py

@@ -70,6 +70,11 @@ def shutdown(self) -> None:
         """Called when a :class:`opentelemetry.sdk.trace.Tracer` is shutdown.
         """
 
+    def force_flush(self) -> None:
+        """Export all ended spans to the configured Exporter that have not
+        yet been exported.
+        """
+
 
 class MultiSpanProcessor(SpanProcessor):
     """Implementation of :class:`SpanProcessor` that forwards all received

opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py

@@ -83,6 +83,9 @@ def on_end(self, span: Span) -> None:
     def shutdown(self) -> None:
         self.span_exporter.shutdown()
 
+    def force_flush(self) -> None:
+        pass
+
 
 class BatchExportSpanProcessor(SpanProcessor):
     """Batch span processor implementation.
@@ -171,7 +174,7 @@ def worker(self):
             timeout = self.schedule_delay_millis / 1e3 - duration
 
         # be sure that all spans are sent
-        self._flush()
+        self.force_flush()
 
     def export(self) -> None:
         """Exports at most max_export_batch_size spans."""
@@ -197,7 +200,7 @@ def export(self) -> None:
         for index in range(idx):
             self.spans_list[index] = None
 
-    def _flush(self):
+    def force_flush(self):
         # export all elements until queue is empty
         while self.queue:
             self.export()

opentelemetry-sdk/tests/trace/export/test_export.py

@@ -97,7 +97,7 @@ def _create_start_and_end_span(name, span_processor):
 
 
 class TestBatchExportSpanProcessor(unittest.TestCase):
-    def test_batch_span_processor(self):
+    def test_shutdown(self):
         spans_names_list = []
 
         my_exporter = MySpanExporter(destination=spans_names_list)
@@ -109,9 +109,35 @@ def test_batch_span_processor(self):
             _create_start_and_end_span(name, span_processor)
 
         span_processor.shutdown()
+        self.assertTrue(my_exporter.is_shutdown)
+
+        # check that spans are exported without an explicitly call to
+        # force_flush()
         self.assertListEqual(span_names, spans_names_list)
 
-        self.assertTrue(my_exporter.is_shutdown)
+    def test_flush(self):
+        spans_names_list = []
+
+        my_exporter = MySpanExporter(destination=spans_names_list)
+        span_processor = export.BatchExportSpanProcessor(my_exporter)
+
+        span_names0 = ["xxx", "bar", "foo"]
+        span_names1 = ["yyy", "baz", "fox"]
+
+        for name in span_names0:
+            _create_start_and_end_span(name, span_processor)
+
+        span_processor.force_flush()
+        self.assertListEqual(span_names0, spans_names_list)
+
+        # create some more spans to check that span processor still works
+        for name in span_names1:
+            _create_start_and_end_span(name, span_processor)
+
+        span_processor.force_flush()
+        self.assertListEqual(span_names0 + span_names1, spans_names_list)
+
+        span_processor.shutdown()
 
     def test_batch_span_processor_lossless(self):
         """Test that no spans are lost when sending max_queue_size spans"""
@@ -127,8 +153,9 @@ def test_batch_span_processor_lossless(self):
         for _ in range(512):
             _create_start_and_end_span("foo", span_processor)
 
-        span_processor.shutdown()
+        span_processor.force_flush()
         self.assertEqual(len(spans_names_list), 512)
+        span_processor.shutdown()
 
     def test_batch_span_processor_many_spans(self):
         """Test that no spans are lost when sending many spans"""
@@ -150,8 +177,9 @@ def test_batch_span_processor_many_spans(self):
 
             time.sleep(0.05)  # give some time for the exporter to upload spans
 
-        span_processor.shutdown()
+        span_processor.force_flush()
         self.assertEqual(len(spans_names_list), 1024)
+        span_processor.shutdown()
 
     def test_batch_span_processor_scheduled_delay(self):
         """Test that spans are exported each schedule_delay_millis"""