feat(profiling/gevent): supports tracking tasks in gevent < 1.3 (#2929)

This implements a greenlet tracer similar to the one that gevent 1.3 ships. It
is actually simpler than the one gevent provides as we don't need as many
features, so it should be _slightly_ faster.

This makes sure we can provide the same amazing experience whatever the gevent
version being used might be.

Co-authored-by: Kyle Verhoog <[email protected]>
Co-authored-by: Brett Langdon <[email protected]>

Author: 3 people

ddtrace/profiling/collector/_task.pyx

@@ -5,18 +5,29 @@ from . import _threading
 
 
 try:
-    import gevent._tracer
+    from greenlet import settrace, getcurrent
     import gevent.thread
+    import gevent.hub
 except ImportError:
     _gevent_tracer = None
 else:
-    class DDGreenletTracer(gevent._tracer.GreenletTracer):
-        def _trace(self, event, args):
-            # Do not trace gevent Hub: the Hub is a greenlet but we want to know the latest active greenlet *before*
-            # the application yielded back to the Hub. There's no point showing the Hub most of the time to the users as
-            # that does not give any information about user code.
-            if not isinstance(args[1], gevent.hub.Hub):
-                gevent._tracer.GreenletTracer._trace(self, event, args)
+
+    class DDGreenletTracer(object):
+        def __init__(self):
+            # type: (...) -> None
+            self.active_greenlet = getcurrent()
+            self.previous_trace_function = settrace(self)
+
+        def __call__(self, event, args):
+            if event in ('switch', 'throw'):
+                # Do not trace gevent Hub: the Hub is a greenlet but we want to know the latest active greenlet *before*
+                # the application yielded back to the Hub. There's no point showing the Hub most of the time to the
+                # users as that does not give any information about user code.
+                if not isinstance(args[1], gevent.hub.Hub):
+                    self.active_greenlet = args[1]
+
+            if self.previous_trace_function is not None:
+                self.previous_trace_function(event, args)
 
     # NOTE: bold assumption: this module is always imported by the MainThread.
     # A GreenletTracer is local to the thread instantiating it and we assume this is run by the MainThread.
@@ -28,16 +39,10 @@ cpdef get_task(thread_id):
     # gevent greenlet support:
     # we only support tracing tasks in the greenlets are run in the MainThread.
     if thread_id == nogevent.main_thread_id and _gevent_tracer is not None:
-        if _gevent_tracer.active_greenlet is None:
-            # That means gevent never switch to another greenlet, we're still in the main one
-            task_id = compat.main_thread.ident
-            frame = None
-        else:
-            task_id = gevent.thread.get_ident(_gevent_tracer.active_greenlet)
-            frame = _gevent_tracer.active_greenlet.gr_frame
-
+        task_id = gevent.thread.get_ident(_gevent_tracer.active_greenlet)
         # Greenlets might be started as Thread in gevent
         task_name = _threading.get_thread_name(task_id)
+        frame = _gevent_tracer.active_greenlet.gr_frame
     else:
         task_id = None
         task_name = None

releasenotes/notes/profiling-gevent-support-older-8e1eb89e36ba8c14.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Profiling now supports tracing greenlets with gevent version prior to 1.3.

tests/profiling/collector/test_stack.py

@@ -15,7 +15,6 @@
 from ddtrace.profiling import event as event_mod
 from ddtrace.profiling import profiler
 from ddtrace.profiling import recorder
-from ddtrace.profiling.collector import _task
 from ddtrace.profiling.collector import _threading
 from ddtrace.profiling.collector import stack
 
@@ -70,7 +69,7 @@ def test_collect_once():
     stack_events = all_events[0]
     for e in stack_events:
         if e.thread_name == "MainThread":
-            if TESTING_GEVENT and _task._gevent_tracer is not None:
+            if TESTING_GEVENT:
                 assert e.task_id > 0
                 assert e.task_name == e.thread_name
             else:
@@ -95,7 +94,7 @@ def _fib(n):
         return _fib(n - 1) + _fib(n - 2)
 
 
-@pytest.mark.skipif(_task._gevent_tracer is None, reason="gevent tasks not supported")
+@pytest.mark.skipif(not TESTING_GEVENT, reason="Not testing gevent")
 def test_collect_gevent_thread_task():
     r = recorder.Recorder()
     s = stack.StackCollector(r)
@@ -163,7 +162,7 @@ def collect(self):
         return []
 
 
-@pytest.mark.skipif(_task._gevent_tracer is None, reason="gevent tasks not supported")
+@pytest.mark.skipif(not TESTING_GEVENT, reason="Not testing gevent")
 @pytest.mark.parametrize("ignore", (True, False))
 def test_ignore_profiler_gevent_task(monkeypatch, ignore):
     monkeypatch.setenv("DD_PROFILING_API_TIMEOUT", "0.1")
@@ -329,7 +328,7 @@ def test_exception_collection():
     assert e.sampling_period > 0
     assert e.thread_id == nogevent.thread_get_ident()
     assert e.thread_name == "MainThread"
-    assert e.frames == [(__file__, 323, "test_exception_collection")]
+    assert e.frames == [(__file__, 322, "test_exception_collection")]
     assert e.nframes == 1
     assert e.exc_type == ValueError
 
@@ -352,7 +351,7 @@ def test_exception_collection_trace(tracer):
     assert e.sampling_period > 0
     assert e.thread_id == nogevent.thread_get_ident()
     assert e.thread_name == "MainThread"
-    assert e.frames == [(__file__, 346, "test_exception_collection_trace")]
+    assert e.frames == [(__file__, 345, "test_exception_collection_trace")]
     assert e.nframes == 1
     assert e.exc_type == ValueError
     assert e.span_id == span.span_id
@@ -597,7 +596,7 @@ def test_thread_time_cache():
         )
 
 
-@pytest.mark.skipif(_task._gevent_tracer is None, reason="gevent tasks not supported")
+@pytest.mark.skipif(not TESTING_GEVENT, reason="Not testing gevent")
 def test_collect_gevent_thread_hub():
     # type: (...) -> None
     r = recorder.Recorder()

tests/profiling/collector/test_threading.py

@@ -1,16 +1,19 @@
+import os
 import threading
 import uuid
 
 import pytest
 
 from ddtrace.internal import nogevent
 from ddtrace.profiling import recorder
-from ddtrace.profiling.collector import _task
 from ddtrace.profiling.collector import threading as collector_threading
 
 from . import test_collector
 
 
+TESTING_GEVENT = os.getenv("DD_PROFILE_TEST_GEVENT", False)
+
+
 def test_repr():
     test_collector._test_repr(
         collector_threading.LockCollector,
@@ -63,13 +66,13 @@ def test_lock_acquire_events():
     assert len(r.events[collector_threading.LockAcquireEvent]) == 1
     assert len(r.events[collector_threading.LockReleaseEvent]) == 0
     event = r.events[collector_threading.LockAcquireEvent][0]
-    assert event.lock_name == "test_threading.py:61"
+    assert event.lock_name == "test_threading.py:64"
     assert event.thread_id == nogevent.thread_get_ident()
     assert event.wait_time_ns > 0
     # It's called through pytest so I'm sure it's gonna be that long, right?
     assert len(event.frames) > 3
     assert event.nframes > 3
-    assert event.frames[0] == (__file__, 62, "test_lock_acquire_events")
+    assert event.frames[0] == (__file__, 65, "test_lock_acquire_events")
     assert event.sampling_pct == 100
 
 
@@ -90,14 +93,14 @@ def test_lock_events_tracer(tracer):
     events = r.reset()
     # The tracer might use locks, so we need to look into every event to assert we got ours
     for event_type in (collector_threading.LockAcquireEvent, collector_threading.LockReleaseEvent):
-        assert {"test_threading.py:81", "test_threading.py:84"}.issubset({e.lock_name for e in events[event_type]})
+        assert {"test_threading.py:84", "test_threading.py:87"}.issubset({e.lock_name for e in events[event_type]})
         for event in events[event_type]:
-            if event.name == "test_threading.py:81":
+            if event.name == "test_threading.py:84":
                 assert event.trace_id is None
                 assert event.span_id is None
                 assert event.trace_resource_container is None
                 assert event.trace_type is None
-            elif event.name == "test_threading.py:84":
+            elif event.name == "test_threading.py:87":
                 assert event.trace_id == trace_id
                 assert event.span_id == span_id
                 assert event.trace_resource_container[0] == t.resource
@@ -123,14 +126,14 @@ def test_lock_events_tracer_late_finish(tracer):
     events = r.reset()
     # The tracer might use locks, so we need to look into every event to assert we got ours
     for event_type in (collector_threading.LockAcquireEvent, collector_threading.LockReleaseEvent):
-        assert {"test_threading.py:112", "test_threading.py:115"}.issubset({e.lock_name for e in events[event_type]})
+        assert {"test_threading.py:115", "test_threading.py:118"}.issubset({e.lock_name for e in events[event_type]})
         for event in events[event_type]:
-            if event.name == "test_threading.py:112":
+            if event.name == "test_threading.py:115":
                 assert event.trace_id is None
                 assert event.span_id is None
                 assert event.trace_resource_container is None
                 assert event.trace_type is None
-            elif event.name == "test_threading.py:115":
+            elif event.name == "test_threading.py:118":
                 assert event.trace_id == trace_id
                 assert event.span_id == span_id
                 assert event.trace_resource_container[0] == span.resource
@@ -155,14 +158,14 @@ def test_resource_not_collected(monkeypatch, tracer):
     events = r.reset()
     # The tracer might use locks, so we need to look into every event to assert we got ours
     for event_type in (collector_threading.LockAcquireEvent, collector_threading.LockReleaseEvent):
-        assert {"test_threading.py:146", "test_threading.py:149"}.issubset({e.lock_name for e in events[event_type]})
+        assert {"test_threading.py:149", "test_threading.py:152"}.issubset({e.lock_name for e in events[event_type]})
         for event in events[event_type]:
-            if event.name == "test_threading.py:146":
+            if event.name == "test_threading.py:149":
                 assert event.trace_id is None
                 assert event.span_id is None
                 assert event.trace_resource_container is None
                 assert event.trace_type is None
-            elif event.name == "test_threading.py:149":
+            elif event.name == "test_threading.py:152":
                 assert event.trace_id == trace_id
                 assert event.span_id == span_id
                 assert event.trace_resource_container is None
@@ -178,17 +181,17 @@ def test_lock_release_events():
     assert len(r.events[collector_threading.LockAcquireEvent]) == 1
     assert len(r.events[collector_threading.LockReleaseEvent]) == 1
     event = r.events[collector_threading.LockReleaseEvent][0]
-    assert event.lock_name == "test_threading.py:175"
+    assert event.lock_name == "test_threading.py:178"
     assert event.thread_id == nogevent.thread_get_ident()
     assert event.locked_for_ns >= 0.1
     # It's called through pytest so I'm sure it's gonna be that long, right?
     assert len(event.frames) > 3
     assert event.nframes > 3
-    assert event.frames[0] == (__file__, 177, "test_lock_release_events")
+    assert event.frames[0] == (__file__, 180, "test_lock_release_events")
     assert event.sampling_pct == 100
 
 
-@pytest.mark.skipif(_task._gevent_tracer is None, reason="gevent tasks not supported")
+@pytest.mark.skipif(not TESTING_GEVENT, reason="Not testing gevent")
 def test_lock_gevent_tasks():
     r = recorder.Recorder()
 
@@ -205,33 +208,39 @@ def play_with_lock():
     assert len(r.events[collector_threading.LockAcquireEvent]) >= 1
     assert len(r.events[collector_threading.LockReleaseEvent]) >= 1
 
-    event = r.events[collector_threading.LockAcquireEvent][0]
-    assert event.lock_name == "test_threading.py:196"
-    assert event.thread_id == nogevent.main_thread_id
-    assert event.wait_time_ns >= 0
-    assert event.task_id == t.ident
-    assert event.task_name == "foobar"
-    # It's called through pytest so I'm sure it's gonna be that long, right?
-    assert len(event.frames) > 3
-    assert event.nframes > 3
-    assert event.frames[0] == (__file__, 197, "play_with_lock")
-    assert event.sampling_pct == 100
-    assert event.task_id == t.ident
-    assert event.task_name == "foobar"
-
-    event = r.events[collector_threading.LockReleaseEvent][0]
-    assert event.lock_name == "test_threading.py:196"
-    assert event.thread_id == nogevent.main_thread_id
-    assert event.locked_for_ns >= 0.1
-    assert event.task_id == t.ident
-    assert event.task_name == "foobar"
-    # It's called through pytest so I'm sure it's gonna be that long, right?
-    assert len(event.frames) > 3
-    assert event.nframes > 3
-    assert event.frames[0] == (__file__, 198, "play_with_lock")
-    assert event.sampling_pct == 100
-    assert event.task_id == t.ident
-    assert event.task_name == "foobar"
+    for event in r.events[collector_threading.LockAcquireEvent]:
+        if event.lock_name == "test_threading.py:199":
+            assert event.thread_id == nogevent.main_thread_id
+            assert event.wait_time_ns >= 0
+            assert event.task_id == t.ident
+            assert event.task_name == "foobar"
+            # It's called through pytest so I'm sure it's gonna be that long, right?
+            assert len(event.frames) > 3
+            assert event.nframes > 3
+            assert event.frames[0] == (__file__, 200, "play_with_lock")
+            assert event.sampling_pct == 100
+            assert event.task_id == t.ident
+            assert event.task_name == "foobar"
+            break
+    else:
+        pytest.fail("Lock event not found")
+
+    for event in r.events[collector_threading.LockReleaseEvent]:
+        if event.lock_name == "test_threading.py:199":
+            assert event.thread_id == nogevent.main_thread_id
+            assert event.locked_for_ns >= 0.1
+            assert event.task_id == t.ident
+            assert event.task_name == "foobar"
+            # It's called through pytest so I'm sure it's gonna be that long, right?
+            assert len(event.frames) > 3
+            assert event.nframes > 3
+            assert event.frames[0] == (__file__, 201, "play_with_lock")
+            assert event.sampling_pct == 100
+            assert event.task_id == t.ident
+            assert event.task_name == "foobar"
+            break
+    else:
+        pytest.fail("Lock event not found")
 
 
 @pytest.mark.benchmark(