Testbed import

Makefile

@@ -6,7 +6,7 @@ pytest := PYTHONDONTWRITEBYTECODE=1 py.test --tb short -rxs \
 html_report := --cov-report=html
 test_args := --cov-report xml --cov-report term-missing
 
-.PHONY: clean-pyc clean-build docs clean
+.PHONY: clean-pyc clean-build docs clean testbed
 .DEFAULT_GOAL : help
 
 help:
@@ -17,6 +17,7 @@ help:
 	@echo "clean-test - remove test and coverage artifacts"
 	@echo "lint - check style with flake8"
 	@echo "test - run tests quickly with the default Python"
+	@echo "testbed - run testbed scenarios with the default Python"
 	@echo "coverage - check code coverage quickly with the default Python"
 	@echo "docs - generate Sphinx HTML documentation, including API docs"
 	@echo "release - package and upload a release"
@@ -29,6 +30,7 @@ check-virtual-env:
 bootstrap: check-virtual-env
 	pip install -r requirements.txt
 	pip install -r requirements-test.txt
+	pip install -r requirements-testbed.txt
 	python setup.py develop
 
 clean: clean-build clean-pyc clean-test
@@ -57,6 +59,9 @@ lint:
 test:
 	$(pytest) $(test_args)
 
+testbed:
+	python -m testbed
+
 jenkins:
 	pip install -r requirements.txt
 	pip install -r requirements-test.txt

README.rst

@@ -205,6 +205,11 @@ Tests
    make bootstrap
    make test
 
+Testbed suite
+^^^^^^^^^^^^^
+
+A testbed suite designed to test API changes and experimental features is included under the *testbed* directory. For more information, see the `Testbed README <testbed/README.md>`_.
+
 Instrumentation Tests
 ---------------------
 

requirements-testbed.txt

@@ -0,0 +1,5 @@
+# add dependencies in setup.py
+
+-r requirements.txt
+
+-e .[testbed]

setup.py

@@ -35,6 +35,12 @@
             'pytest-mock',
             'Sphinx',
             'sphinx_rtd_theme'
-        ]
+        ],
+        'testbed': [
+            'six>=1.10.0,<2.0',
+            'gevent==1.2',
+            'tornado',
+        ],
+        ':python_version == "2.7"': ['futures'],
     },
 )

testbed/README.md

@@ -0,0 +1,75 @@
+# Testbed suite for the OpenTracing API
+
+Testbed suite designed to test API changes.
+
+## Build and test.
+
+```sh
+make testbed
+```
+
+Depending on whether Python 2 or 3 is being used, the `asyncio` tests will be automatically disabled.
+
+Alternatively, due to the organization of the suite, it's possible to run directly the tests using `py.test`:
+
+```sh
+    py.test -s testbed/test_multiple_callbacks/test_threads.py
+```
+
+## Tested frameworks
+
+Currently the examples cover `threading`, `tornado`, `gevent` and `asyncio` (which requires Python 3). The implementation of `ScopeManager` for each framework is a basic, simple one, and can be found in [span_propagation.py](span_propagation.py). See details below.
+
+### threading
+
+`ThreadScopeManager` uses thread-local storage (through `threading.local()`), and does not provide automatic propagation from thread to thread, which needs to be done manually.
+
+### gevent
+
+`GeventScopeManager` uses greenlet-local storage (through `gevent.local.local()`), and does not provide automatic propagation from parent greenlets to their children, which needs to be done manually.
+
+### tornado
+
+`TornadoScopeManager` uses a variation of `tornado.stack_context.StackContext` to both store **and** automatically propagate the context from parent coroutines to their children. 
+
+Because of this, in order to make the `TornadoScopeManager` work, calls need to be started like this:
+
+```python
+with tracer_stack_context():
+    my_coroutine()
+```
+
+At the moment of writing this, yielding over multiple children is not supported, as the context is effectively shared, and switching from coroutine to coroutine messes up the current active `Span`.
+
+### asyncio
+
+`AsyncioScopeManager` uses the current `Task` (through `Task.current_task()`) to store the active `Span`, and does not provide automatic propagation from parent `Task` to their children, which needs to be done manually.
+
+## List of patterns
+
+- [Active Span replacement](test_active_span_replacement) - Start an isolated task and query for its results in another task/thread.
+- [Client-Server](test_client_server) - Typical client-server example.
+- [Common Request Handler](test_common_request_handler) - One request handler for all requests.
+- [Late Span finish](test_late_span_finish) - Late parent `Span` finish.
+- [Multiple callbacks](test_multiple_callbacks) - Multiple callbacks spawned at the same time.
+- [Nested callbacks](test_nested_callbacks) - One callback at a time, defined ina pipeline fashion.
+- [Subtask Span propagation](test_subtask_span_propagation) - `Span` propagation for subtasks/coroutines.
+
+## Adding new patterns
+
+A new pattern is composed of a directory under *testbed* with the *test_* prefix, and containing the files for each platform, also with the *test_* prefix:
+
+```
+testbed/
+  test_new_pattern/
+    test_threads.py
+    test_tornado.py
+    test_asyncio.py
+    test_gevent.py
+```
+
+Supporting all the platforms is optional, and a warning will be displayed when doing `make testbed` in such case.
+
+## Flake8 support
+
+Currently `flake8` does not support the Python 3 `await`/`async` syntax, and does not offer a way to ignore such syntax.

testbed/__init__.py

@@ -0,0 +1 @@
+

testbed/__main__.py

@@ -0,0 +1,50 @@
+from importlib import import_module
+import logging
+import os
+import six
+import unittest
+
+
+enabled_platforms = [
+    'threads',
+    'tornado',
+    'gevent',
+]
+if six.PY3:
+    enabled_platforms.append('asyncio')
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__package__)
+
+
+def import_test_module(test_name, platform):
+    full_path = '%s.%s.test_%s' % (__package__, test_name, platform)
+    try:
+        return import_module(full_path)
+    except ImportError:
+        pass
+
+    return None
+
+
+def get_test_directories():
+    """Return all the directories starting with test_ under this package."""
+    return [directory for directory in os.listdir(os.path.dirname(__file__))
+            if directory.startswith('test_')]
+
+
+main_suite = unittest.TestSuite()
+loader = unittest.TestLoader()
+
+for test_dir in get_test_directories():
+    for platform in enabled_platforms:
+
+        test_module = import_test_module(test_dir, platform)
+        if test_module is None:
+            logger.warning('Could not load %s for %s' % (test_dir, platform))
+            continue
+
+        suite = loader.loadTestsFromModule(test_module)
+        main_suite.addTests(suite)
+
+unittest.TextTestRunner(verbosity=3).run(main_suite)

testbed/span_propagation.py

@@ -0,0 +1,158 @@
+import six
+import threading
+from tornado.stack_context import StackContext
+import gevent.local
+
+from opentracing import ScopeManager, Scope
+
+if six.PY3:
+    import asyncio
+
+
+#
+# asyncio section.
+#
+class AsyncioScopeManager(ScopeManager):
+    def activate(self, span, finish_on_close):
+        scope = AsyncioScope(self, span, finish_on_close)
+
+        loop = asyncio.get_event_loop()
+        task = asyncio.Task.current_task(loop=loop)
+        setattr(task, '__active', scope)
+
+        return scope
+
+    def _get_current_task(self):
+        loop = asyncio.get_event_loop()
+        return asyncio.Task.current_task(loop=loop)
+
+    @property
+    def active(self):
+        task = self._get_current_task()
+        return getattr(task, '__active', None)
+
+
+class AsyncioScope(Scope):
+    def __init__(self, manager, span, finish_on_close):
+        super(AsyncioScope, self).__init__(manager, span)
+        self._finish_on_close = finish_on_close
+        self._to_restore = manager.active
+
+    def close(self):
+        if self.manager.active is not self:
+            return
+
+        task = self.manager._get_current_task()
+        setattr(task, '__active', self._to_restore)
+
+        if self._finish_on_close:
+            self.span.finish()
+
+#
+# gevent section.
+#
+class GeventScopeManager(ScopeManager):
+    def __init__(self):
+        self._locals = gevent.local.local()
+
+    def activate(self, span, finish_on_close):
+        scope = GeventScope(self, span, finish_on_close)
+        setattr(self._locals, 'active', scope)
+
+        return scope
+
+    @property
+    def active(self):
+        return getattr(self._locals, 'active', None)
+
+
+class GeventScope(Scope):
+    def __init__(self, manager, span, finish_on_close):
+        super(GeventScope, self).__init__(manager, span)
+        self._finish_on_close = finish_on_close
+        self._to_restore = manager.active
+
+    def close(self):
+        if self.manager.active is not self:
+            return
+
+        setattr(self.manager._locals, 'active', self._to_restore)
+
+        if self._finish_on_close:
+            self.span.finish()
+
+#
+# tornado section.
+#
+class TornadoScopeManager(ScopeManager):
+    def activate(self, span, finish_on_close):
+        context = self._get_context()
+        if context is None:
+            raise Exception('No StackContext detected')
+
+        scope = TornadoScope(self, span, finish_on_close)
+        context.active = scope
+
+        return scope
+
+    def _get_context(self):
+        return TracerRequestContextManager.current_context()
+
+    @property
+    def active(self):
+        context = self._get_context()
+        if context is None:
+            return None
+
+        return context.active
+
+
+class TornadoScope(Scope):
+    def __init__(self, manager, span, finish_on_close):
+        super(TornadoScope, self).__init__(manager, span)
+        self._finish_on_close = finish_on_close
+        self._to_restore = manager.active
+
+    def close(self):
+        context = self.manager._get_context()
+        if context is None or context.active is not self:
+            return
+
+        context.active = self._to_restore
+
+        if self._finish_on_close:
+            self.span.finish()
+
+
+class TracerRequestContext(object):
+    __slots__ = ('active', )
+
+    def __init__(self, active=None):
+        self.active = active
+
+
+class TracerRequestContextManager(object):
+    _state = threading.local()
+    _state.context = None
+
+    @classmethod
+    def current_context(cls):
+        return getattr(cls._state, 'context', None)
+
+    def __init__(self, context):
+        self._context = context
+
+    def __enter__(self):
+        self._prev_context = self.__class__.current_context()
+        self.__class__._state.context = self._context
+        return self._context
+
+    def __exit__(self, *_):
+        self.__class__._state.context = self._prev_context
+        self._prev_context = None
+        return False
+
+
+def tracer_stack_context():
+    context = TracerRequestContext()
+    return StackContext(lambda: TracerRequestContextManager(context))

testbed/test_active_span_replacement/README.md

@@ -0,0 +1,18 @@
+# Active Span replacement example.
+
+This example shows a `Span` being created and then passed to an asynchronous task, which will temporary activate it to finish its processing, and further restore the previously active `Span`.
+
+`threading` implementation:
+```python
+# Create a new Span for this task
+with self.tracer.start_active_span('task'):
+
+    with self.tracer.scope_manager.activate(span, True):
+	# Simulate work strictly related to the initial Span
+	pass
+
+    # Use the task span as parent of a new subtask
+    with self.tracer.start_active_span('subtask'):
+	pass
+
+```