sdk: Implement force_flush for span processors

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"""