Make wait_period configurable in async transport (#618)

Make wait_period configurable for AsyncTransport.
Clean up the inline docs.
Switch Azure exporter to AsyncTransport.

Author: reyang

contrib/opencensus-ext-azure/opencensus/ext/azure/trace_exporter/__init__.py

@@ -15,7 +15,7 @@
 import json
 import requests
 
-from opencensus.common.transports import sync
+from opencensus.common.transports.async_ import AsyncTransport
 from opencensus.ext.azure.common import Options
 from opencensus.ext.azure.common import utils
 from opencensus.ext.azure.common.protocol import Data
@@ -26,24 +26,19 @@
 from opencensus.trace import execution_context
 from opencensus.trace.span import SpanKind
 
+__all__ = ['AzureExporter']
+
 
 class AzureExporter(base_exporter.Exporter):
     """An exporter that sends traces to Microsoft Azure Monitor.
 
     :type options: dict
     :param options: Options for the exporter. Defaults to None.
-
-    :type transport: :class:`type`
-    :param transport: Class for creating new transport objects. It should
-                      extend from the base_exporter :class:`.Transport` type
-                      and implement :meth:`.Transport.export`. Defaults to
-                      :class:`.SyncTransport`. The other option is
-                      :class:`.AsyncTransport`.
     """
 
-    def __init__(self, options=None, transport=sync.SyncTransport):
+    def __init__(self, options=None):
         self.options = options or Options()
-        self.transport = transport(self)
+        self.transport = AsyncTransport(self, max_batch_size=100)
 
     def span_data_to_envelope(self, sd):
         # print('[AzMon]', sd)

contrib/opencensus-ext-azure/tests/test_azure_trace_exporter.py

@@ -19,13 +19,15 @@
 
 class TestAzureExporter(unittest.TestCase):
     def test_export(self):
-        exporter = trace_exporter.AzureExporter(transport=MockTransport)
+        exporter = trace_exporter.AzureExporter()
+        exporter.transport = MockTransport()
         exporter.export(None)
         self.assertTrue(exporter.transport.export_called)
 
     @mock.patch('requests.post', return_value=mock.Mock())
     def test_emit(self, request_mock):
-        exporter = trace_exporter.AzureExporter(transport=MockTransport)
+        exporter = trace_exporter.AzureExporter()
+        exporter.transport = MockTransport()
         exporter.emit([])
 
     def test_span_data_to_envelope(self):

opencensus/common/transports/async_.py

@@ -23,7 +23,7 @@
 
 _DEFAULT_GRACE_PERIOD = 5.0  # Seconds
 _DEFAULT_MAX_BATCH_SIZE = 600
-_WAIT_PERIOD = 60.0  # Seconds
+_DEFAULT_WAIT_PERIOD = 60.0  # Seconds
 _WORKER_THREAD_NAME = 'opencensus.common.Worker'
 _WORKER_TERMINATOR = object()
 
@@ -33,10 +33,7 @@ class _Worker(object):
 
     :type exporter: :class:`~opencensus.trace.base_exporter.Exporter` or
                     :class:`~opencensus.stats.base_exporter.StatsExporter`
-    :param exporter: Instances of Exporter objects. Defaults to
-                    :class:`.PrintExporter`. The rest options are
-                    :class:`.ZipkinExporter`, :class:`.StackdriverExporter`,
-                    :class:`.LoggingExporter`, :class:`.FileExporter`.
+    :param exporter: Instance of Exporter object.
 
     :type grace_period: float
     :param grace_period: The amount of time to wait for pending data to
@@ -45,12 +42,19 @@ class _Worker(object):
     :type max_batch_size: int
     :param max_batch_size: The maximum number of items to send at a time
                            in the background thread.
+
+    :type wait_period: int
+    :param wait_period: The amount of time to wait before sending the next
+                        batch of data.
     """
-    def __init__(self, exporter, grace_period=_DEFAULT_GRACE_PERIOD,
-                 max_batch_size=_DEFAULT_MAX_BATCH_SIZE):
+    def __init__(self, exporter,
+                 grace_period=_DEFAULT_GRACE_PERIOD,
+                 max_batch_size=_DEFAULT_MAX_BATCH_SIZE,
+                 wait_period=_DEFAULT_WAIT_PERIOD):
         self.exporter = exporter
         self._grace_period = grace_period
         self._max_batch_size = max_batch_size
+        self._wait_period = wait_period
         self._queue = queue.Queue(0)
         self._lock = threading.Lock()
         self._event = threading.Event()
@@ -116,8 +120,8 @@ def _thread_main(self):
 
             # self._event is set at exit, at which point we start draining the
             # queue immediately. If self._event is unset, block for
-            # _WAIT_PERIOD between each batch of exports.
-            self._event.wait(_WAIT_PERIOD)
+            # self.wait_period between each batch of exports.
+            self._event.wait(self._wait_period)
 
             if quit_:
                 break
@@ -185,10 +189,7 @@ class AsyncTransport(base.Transport):
 
     :type exporter: :class:`~opencensus.trace.base_exporter.Exporter` or
                     :class:`~opencensus.stats.base_exporter.StatsExporter`
-    :param exporter: Instances of Exporter objects. Defaults to
-                     :class:`.PrintExporter`. The rest options are
-                     :class:`.ZipkinExporter`, :class:`.StackdriverExporter`,
-                     :class:`.LoggingExporter`, :class:`.FileExporter`.
+    :param exporter: Instance of Exporter object.
 
     :type grace_period: float
     :param grace_period: The amount of time to wait for pending data to
@@ -197,12 +198,23 @@ class AsyncTransport(base.Transport):
     :type max_batch_size: int
     :param max_batch_size: The maximum number of items to send at a time
                            in the background thread.
+
+    :type wait_period: int
+    :param wait_period: The amount of time to wait before sending the next
+                        batch of data.
     """
 
-    def __init__(self, exporter, grace_period=_DEFAULT_GRACE_PERIOD,
-                 max_batch_size=_DEFAULT_MAX_BATCH_SIZE):
+    def __init__(self, exporter,
+                 grace_period=_DEFAULT_GRACE_PERIOD,
+                 max_batch_size=_DEFAULT_MAX_BATCH_SIZE,
+                 wait_period=_DEFAULT_WAIT_PERIOD):
         self.exporter = exporter
-        self.worker = _Worker(exporter, grace_period, max_batch_size)
+        self.worker = _Worker(
+            exporter,
+            grace_period,
+            max_batch_size,
+            wait_period,
+        )
         self.worker.start()
 
     def export(self, data):

tests/unit/common/transports/test_async.py

@@ -19,11 +19,6 @@
 from opencensus.common.transports import async_
 
 
-# Don't let workers wait between exports in testing
-wait_period_patch = mock.patch(
-    'opencensus.common.transports.async_._WAIT_PERIOD', 0)
-
-
 class Test_Worker(unittest.TestCase):
 
     def _start_worker(self, worker):
@@ -118,7 +113,7 @@ def test__export_pending_data_did_not_join(self):
 
     def test__thread_main(self):
         exporter = mock.Mock()
-        worker = async_._Worker(exporter)
+        worker = async_._Worker(exporter, wait_period=0)
 
         trace1 = {
             'traceId': 'test1',
@@ -133,15 +128,14 @@ def test__thread_main(self):
         worker.enqueue(trace2)
         worker._queue.put_nowait(async_._WORKER_TERMINATOR)
 
-        with wait_period_patch:
-            worker._thread_main()
+        worker._thread_main()
 
         self.assertTrue(worker.exporter.emit.called)
         self.assertEqual(worker._queue.qsize(), 0)
 
     def test__thread_main_batches(self):
         exporter = mock.Mock()
-        worker = async_._Worker(exporter, max_batch_size=2)
+        worker = async_._Worker(exporter, max_batch_size=2, wait_period=0)
 
         # Enqueue three records and the termination signal. This should be
         # enough to perform two separate batches and a third loop with just
@@ -169,8 +163,7 @@ def test__thread_main_batches(self):
 
         worker._queue.put_nowait(async_._WORKER_TERMINATOR)
 
-        with wait_period_patch:
-            worker._thread_main()
+        worker._thread_main()
 
         self.assertEqual(worker._queue.qsize(), 0)
 
@@ -184,7 +177,7 @@ def emit(self, span):
                 self.exported.append(span)
 
         exporter = Exporter()
-        worker = async_._Worker(exporter, max_batch_size=2)
+        worker = async_._Worker(exporter, max_batch_size=2, wait_period=0)
 
         # Enqueue three records and the termination signal. This should be
         # enough to perform two separate batches and a third loop with just
@@ -199,8 +192,7 @@ def emit(self, span):
         worker.enqueue(span_data1)
         worker.enqueue(span_data2)
 
-        with wait_period_patch:
-            worker._thread_main()
+        worker._thread_main()
 
         self.assertEqual(exporter.exported, [span_data1])
 
@@ -221,7 +213,7 @@ def emit(self, span):
                     raise Exception("This exporter is broken !")
 
         exporter = Exporter()
-        worker = async_._Worker(exporter, max_batch_size=2)
+        worker = async_._Worker(exporter, max_batch_size=2, wait_period=0)
 
         span_data0 = [mock.Mock()]
         span_data1 = [mock.Mock()]
@@ -232,8 +224,7 @@ def emit(self, span):
         worker.enqueue(span_data2)
         worker.enqueue(async_._WORKER_TERMINATOR)
 
-        with wait_period_patch:
-            worker._thread_main()
+        worker._thread_main()
 
         # Span 2 should throw an exception, only span 0 and 1 are left
         self.assertEqual(exporter.exported, span_data0 + span_data1)