Merge pull request #74 from DataDog/talwai/gevent

[trace] add gevent buffer and docs

Author: talwai

ddtrace/buffer.py

@@ -12,8 +12,8 @@ def get(self):
         raise NotImplementedError()
 
 
-class ThreadLocalSpanBuffer(object):
-    """ ThreadLocalBuffer stores the current active span in thread-local
+class ThreadLocalSpanBuffer(SpanBuffer):
+    """ ThreadLocalSpanBuffer stores the current active span in thread-local
         storage.
     """
 

ddtrace/contrib/gevent/__init__.py

@@ -0,0 +1,76 @@
+"""
+To trace a request in a gevent-ed environment, configure the tracer to use greenlet-local
+storage, rather than the default thread-local storage.
+
+This allows the tracer to pick up a transaction exactly
+where it left off as greenlets yield context to one another.
+
+The simplest way to trace with greenlet-local storage is via the `gevent.monkey` module::
+
+    # Always monkey patch before importing the global tracer
+    # Broadly, gevent recommends that patches happen as early as possible in the app lifecycle
+    # http://www.gevent.org/gevent.monkey.html#patching
+
+    from gevent import monkey; monkey.patch_thread()
+    # Alternatively, use monkey.patch_all() to perform all available patches
+
+    from ddtrace import tracer
+
+    import gevent
+
+    def my_parent_function():
+        with tracer.trace("web.request") as span:
+            span.service = "web"
+            gevent.spawn(worker_function, span)
+
+    def worker_function(parent):
+        # Set the active span
+        tracer.span_buffer.set(parent)
+
+        # then trace its child
+        with tracer.trace("greenlet.call") as span:
+            span.service = "greenlet"
+            ...
+
+            with tracer.trace("greenlet.child_call") as child:
+                ...
+
+Note that when spawning greenlets,
+the span object must be explicitly passed from the parent to coroutine context.
+A tracer in a freshly-spawned greenlet will not know about its parent span.
+
+If you are unable to patch `gevent` in the global scope, you can configure
+the global tracer to use greenlet-local storage on an as-needed basis::
+
+    from ddtrace import tracer
+    from ddtrace.contrib.gevent import GreenletLocalSpanBuffer
+
+    import gevent
+
+    def my_parent_function():
+        with tracer.trace("web.request") as span:
+            span.service = "web"
+            gevent.spawn(worker_function, span)
+
+    def worker_function(parent):
+        tracer.span_buffer = GreenletLocalSpanBuffer()
+        # Set the active span
+        tracer.span_buffer.set(parent)
+
+        # then trace its child
+        with tracer.trace("greenlet.call") as span:
+            span.service = "greenlet"
+            ...
+
+            with tracer.trace("greenlet.child_call") as child:
+                ...
+"""
+
+from ..util import require_modules
+
+required_modules = ['gevent', 'gevent.local']
+
+with require_modules(required_modules) as missing_modules:
+    if not missing_modules:
+        from .buffer import GreenletLocalSpanBuffer
+        __all__ = ['GreenletLocalSpanBuffer']

ddtrace/contrib/gevent/buffer.py

@@ -0,0 +1,21 @@
+import gevent.local
+from ddtrace.buffer import SpanBuffer
+
+class GreenletLocalSpanBuffer(SpanBuffer):
+    """ GreenletLocalSpanBuffer stores the current active span in greenlet-local
+        storage.
+    """
+
+    def __init__(self):
+        self._locals = gevent.local.local()
+
+    def set(self, span):
+        self._locals.span = span
+
+    def get(self):
+        return getattr(self._locals, 'span', None)
+
+    def pop(self):
+        span = self.get()
+        self.set(None)
+        return span

ddtrace/contrib/sqlalchemy/__init__.py

@@ -10,6 +10,15 @@
     trace_engine(engine, tracer, "my-database")
 
     engine.connect().execute("select count(*) from users")
+
+If you are using sqlalchemy in a gevent-ed environment, make sure to monkey patch
+the `thread` module prior to importing the global tracer::
+
+    from gevent import monkey; monkey.patch_thread() # or patch_all() if you prefer
+    from ddtrace import tracer
+
+    # Add instrumentation to your engine as above
+    ...
 """
 
 

docs/index.rst

@@ -148,6 +148,12 @@ Flask-cache
 
 .. automodule:: ddtrace.contrib.flask_cache
 
+
+Gevent
+~~~~~~
+
+.. automodule:: ddtrace.contrib.gevent
+
 MongoDB
 ~~~~~~~
 

tests/contrib/gevent/test.py

@@ -0,0 +1,117 @@
+import unittest
+
+from nose.tools import eq_, ok_
+from nose.plugins.attrib import attr
+import gevent
+import gevent.local
+import thread
+import threading
+
+
+class GeventGlobalScopeTest(unittest.TestCase):
+    def setUp(self):
+        # simulate standard app bootstrap
+        from gevent import monkey; monkey.patch_thread()
+        from ddtrace import tracer
+
+    def test_global_patch(self):
+        from ddtrace import tracer; tracer.enabled = False
+
+        # Ensure the patch is active
+        ok_(isinstance(tracer.span_buffer._locals, gevent.local.local))
+
+        seen_resources = []
+        def worker_function(parent):
+            tracer.span_buffer.set(parent)
+            seen_resources.append(tracer.span_buffer.get().resource)
+
+            with tracer.trace("greenlet.call") as span:
+                span.resource = "sibling"
+
+                gevent.sleep()
+
+                # Ensure we have the correct parent span even after a context switch
+                eq_(tracer.span_buffer.get().span_id, span.span_id)
+                with tracer.trace("greenlet.other_call") as child:
+                    child.resource = "sibling_child"
+
+        with tracer.trace("web.request") as span:
+            span.service = "web"
+            span.resource = "parent"
+            worker_count = 5
+            workers = [gevent.spawn(worker_function, span) for w in range(worker_count)]
+            gevent.joinall(workers)
+
+        # Ensure all greenlets see the right parent span
+        ok_("sibling" not in seen_resources)
+        ok_(all(s == "parent" for s in seen_resources))
+
+    def tearDown(self):
+        # undo gevent monkey patching
+        reload(thread); reload(threading)
+        from ddtrace.buffer import ThreadLocalSpanBuffer
+        from ddtrace import tracer; tracer.span_buffer = ThreadLocalSpanBuffer()
+
+
+class GeventLocalScopeTest(unittest.TestCase):
+
+    def test_unpatched(self):
+        """
+        Demonstrate a situation where thread-local storage leads to a bad tree:
+            1. Main thread spawns several coroutines
+            2. A coroutine is handed context from a sibling coroutine
+            3. A coroutine incorrectly sees a "sibling" span as its parent
+        """
+        from ddtrace import tracer; tracer.enabled = False
+
+        seen_resources = []
+        def my_worker_function(i):
+            ok_(tracer.span_buffer.get())
+            seen_resources.append(tracer.span_buffer.get().resource)
+
+            with tracer.trace("greenlet.call") as span:
+                span.resource = "sibling"
+                gevent.sleep()
+
+        with tracer.trace("web.request") as span:
+            span.service = "web"
+            span.resource = "parent"
+
+            worker_count = 5
+            workers = [gevent.spawn(my_worker_function, w) for w in range(worker_count)]
+            gevent.joinall(workers)
+
+        # check that a bad parent span was seen
+        ok_("sibling" in seen_resources)
+
+    def test_local_patch(self):
+        """
+        Test patching a parent span into a coroutine's tracer
+        """
+        from ddtrace import tracer; tracer.enabled = False
+        from ddtrace.contrib.gevent import GreenletLocalSpanBuffer
+
+        def fn(parent):
+            tracer.span_buffer = GreenletLocalSpanBuffer()
+            tracer.span_buffer.set(parent)
+
+            with tracer.trace("greenlet.call") as span:
+                span.service = "greenlet"
+
+                gevent.sleep()
+
+                # Ensure we have the correct parent span even after a context switch
+                eq_(tracer.span_buffer.get().span_id, span.span_id)
+                with tracer.trace("greenlet.child_call") as child:
+                    eq_(child.parent_id, span.span_id)
+
+        with tracer.trace("web.request") as span:
+            span.service = "web"
+            worker = gevent.spawn(fn, span)
+            worker.join()
+
+    def tearDown(self):
+        # undo gevent monkey patching
+        reload(thread); reload(threading)
+        from ddtrace.buffer import ThreadLocalSpanBuffer
+        from ddtrace import tracer; tracer.span_buffer = ThreadLocalSpanBuffer()

tox.ini

@@ -14,6 +14,7 @@ envlist =
     {py27,py34}-django{18,19,110}-djangopylibmc06-djangoredis45-pylibmc-redis-memcached
     {py27,py34}-flask{010,011}-blinker
     {py27,py34}-flask{010,011}-flaskcache{013}-memcached-redis-blinker
+    {py27,py34}-gevent{10,11}
     {py27}-flask{010,011}-flaskcache{012}-memcached-redis-blinker
     {py27,py34}-mysqlconnector{21}
     {py27,py34}-pylibmc{140,150}
@@ -57,6 +58,8 @@ deps =
     djangoredis45: django-redis>=4.5,<4.6
     flask010: flask>=0.10,<0.11
     flask011: flask>=0.11
+    gevent10: gevent>=1.0,<1.1
+    gevent11: gevent>=1.1
     flaskcache012: flask_cache>=0.12,<0.13
     flaskcache013: flask_cache>=0.13,<0.14
     memcached: python-memcached