Fix BatchExportSpanProcessor not resetting timeout on worker loop (open-telemetry#1218)

p-yang · web-flow · commit 97b602fe7d77 · 2020-10-27T15:46:19.000-07:00
diff --git a/exporter/opentelemetry-exporter-datadog/src/opentelemetry/exporter/datadog/spanprocessor.py b/exporter/opentelemetry-exporter-datadog/src/opentelemetry/exporter/datadog/spanprocessor.py
@@ -119,7 +119,8 @@ def worker(self):
                 with self.condition:
                     self.condition.wait(timeout)
                     if not self.check_traces_queue:
-                        # spurious notification, let's wait again
+                        # spurious notification, let's wait again, reset timeout
+                        timeout = self.schedule_delay_millis / 1e3
                         continue
                     if self.done:
                         # missing spans will be sent when calling flush
diff --git a/exporter/opentelemetry-exporter-datadog/tests/test_datadog_exporter.py b/exporter/opentelemetry-exporter-datadog/tests/test_datadog_exporter.py
@@ -483,6 +483,42 @@ def test_span_processor_scheduled_delay(self):
 
         tracer_provider.shutdown()
 
+    def test_batch_span_processor_reset_timeout(self):
+        """Test that the scheduled timeout is reset on cycles without spans"""
+        delay = 50
+        # pylint: disable=protected-access
+        exporter = MockDatadogSpanExporter()
+        exporter._agent_writer.write.side_effect = lambda spans: time.sleep(
+            0.05
+        )
+        span_processor = datadog.DatadogExportSpanProcessor(
+            exporter, schedule_delay_millis=delay
+        )
+        tracer_provider = trace.TracerProvider()
+        tracer_provider.add_span_processor(span_processor)
+        tracer = tracer_provider.get_tracer(__name__)
+        with mock.patch.object(span_processor.condition, "wait") as mock_wait:
+            with tracer.start_span("foo"):
+                pass
+
+            # give some time for exporter to loop
+            # since wait is mocked it should return immediately
+            time.sleep(0.1)
+            mock_wait_calls = list(mock_wait.mock_calls)
+
+            # find the index of the call that processed the singular span
+            for idx, wait_call in enumerate(mock_wait_calls):
+                _, args, __ = wait_call
+                if args[0] <= 0:
+                    after_calls = mock_wait_calls[idx + 1 :]
+                    break
+
+            self.assertTrue(
+                all(args[0] >= 0.05 for _, args, __ in after_calls)
+            )
+
+        span_processor.shutdown()
+
     def test_span_processor_accepts_parent_context(self):
         span_processor = mock.Mock(
             wraps=datadog.DatadogExportSpanProcessor(self.exporter)
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/trace/export/__init__.py
@@ -213,7 +213,8 @@ def worker(self):
                     self.condition.wait(timeout)
                     flush_request = self._get_and_unset_flush_request()
                     if not self.queue:
-                        # spurious notification, let's wait again
+                        # spurious notification, let's wait again, reset timeout
+                        timeout = self.schedule_delay_millis / 1e3
                         self._notify_flush_request_finished(flush_request)
                         flush_request = None
                         continue
diff --git a/opentelemetry-sdk/tests/trace/export/test_export.py b/opentelemetry-sdk/tests/trace/export/test_export.py
@@ -381,6 +381,43 @@ def test_batch_span_processor_scheduled_delay(self):
 
         span_processor.shutdown()
 
+    def test_batch_span_processor_reset_timeout(self):
+        """Test that the scheduled timeout is reset on cycles without spans"""
+        spans_names_list = []
+
+        export_event = threading.Event()
+        my_exporter = MySpanExporter(
+            destination=spans_names_list,
+            export_event=export_event,
+            export_timeout_millis=50,
+        )
+
+        span_processor = export.BatchExportSpanProcessor(
+            my_exporter, schedule_delay_millis=50,
+        )
+
+        with mock.patch.object(span_processor.condition, "wait") as mock_wait:
+            _create_start_and_end_span("foo", span_processor)
+            self.assertTrue(export_event.wait(2))
+
+            # give some time for exporter to loop
+            # since wait is mocked it should return immediately
+            time.sleep(0.05)
+            mock_wait_calls = list(mock_wait.mock_calls)
+
+            # find the index of the call that processed the singular span
+            for idx, wait_call in enumerate(mock_wait_calls):
+                _, args, __ = wait_call
+                if args[0] <= 0:
+                    after_calls = mock_wait_calls[idx + 1 :]
+                    break
+
+            self.assertTrue(
+                all(args[0] >= 0.05 for _, args, __ in after_calls)
+            )
+
+        span_processor.shutdown()
+
     def test_batch_span_processor_parameters(self):
         # zero max_queue_size
         self.assertRaises(
