fix(profiling): accounting of time for gevent tasks [backport #5338 to 1.11] (#5549)

Backport of #5338 to 1.11

This change fixes the accounting of wall and CPU time to gevent tasks by
only assigning them to the main Python thread, where they are supposed
to run under normal circumstances. This change also ensures that task
information does not leak onto any other threads that might be running,
most notably the profiler threads themselves.

The rough picture of the gevent task modelling on top of Python threads
that comes with this PR is the following


![image](https://user-images.githubusercontent.com/20231758/226345502-3c1cb6a3-0b37-47b3-b711-728647ad6ad9.png)

This means that the `MainThread` will account for the wall time of all
the tasks running within it, as well as the native thread stacks. For
example, if the main, gevent-based, application runs a `sleep(2)` in the
main thread, and a `sleep(1)` in a secondary thread, the `MainThread`
thread would report a total of about 5 seconds of wall time (both of the
sleeps, plus the 2 seconds spent in the gevent hub as part of the native
thread stack), whereas the `MainThread` _task_ would only account for
the 2 seconds of the `sleep(2)`.

<img width="1050" alt="Screenshot 2023-03-20 at 15 35 13"
src="https://user-images.githubusercontent.com/20231758/226390338-bc47941f-3536-4b95-a50a-da0f37a3acff.png">

<img width="1050" alt="Screenshot 2023-03-20 at 15 35 34"
src="https://user-images.githubusercontent.com/20231758/226390409-5091beda-7815-4c1d-a437-0a2a3d9e727e.png">

Some of the existing tests have been adapted to check for the accounting
proposed by this PR. Dedicated scenarios will be added to internal
correctness check to catch future regressions.

## Checklist

- [x] Change(s) are motivated and described in the PR description.
- [x] Testing strategy is described if automated tests are not included
in the PR.
- [x] Risk is outlined (performance impact, potential for breakage,
maintainability, etc).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] [Library release note
guidelines](https://ddtrace.readthedocs.io/en/stable/contributing.html#Release-Note-Guidelines)
are followed.
- [x] Documentation is included (in-code, generated user docs, [public
corp docs](https://github.com/DataDog/documentation/)).
- [x] Author is aware of the performance implications of this PR as
reported in the benchmarks PR comment.

## Reviewer Checklist

- [x] Title is accurate.
- [x] No unnecessary changes are introduced.
- [x] Description motivates each change.
- [x] Avoids breaking
[API](https://ddtrace.readthedocs.io/en/stable/versioning.html#interfaces)
changes unless absolutely necessary.
- [x] Testing strategy adequately addresses listed risk(s).
- [x] Change is maintainable (easy to change, telemetry, documentation).
- [x] Release note makes sense to a user of the library.
- [x] Reviewer is aware of, and discussed the performance implications
of this PR as reported in the benchmarks PR comment.

Author: P403n1x87

ddtrace/__init__.py

@@ -3,7 +3,14 @@
 
 ModuleWatchdog.install()
 
-from ._logger import configure_ddtrace_logger  # noqa: E402
+# Acquire a reference to the threading module. Some parts of the library (e.g.
+# the profiler) might be enabled programmatically and therefore might end up
+# getting a reference to the tracee's threading module. By storing a reference
+# to the threading module used by ddtrace here, we make it easy for those parts
+# to get a reference to the right threading module.
+import threading as _threading
+
+from ._logger import configure_ddtrace_logger
 
 
 # configure ddtrace logger before other modules log

ddtrace/profiling/_threading.pyx

@@ -1,20 +1,18 @@
 from __future__ import absolute_import
 
 import sys
-import threading as ddtrace_threading
 import typing
 import weakref
 
 import attr
 from six.moves import _thread
 
+from ddtrace import _threading as ddtrace_threading
 
-cpdef get_thread_name(thread_id):
-    # Do not force-load the threading module if it's not already loaded
-    if "threading" not in sys.modules:
-        return None
 
-    import threading
+cpdef get_thread_by_id(thread_id):
+    # Do not force-load the threading module if it's not already loaded
+    threading = sys.modules.get("threading", ddtrace_threading)
 
     # Look for all threads, including the ones we create
     for threading_mod in (threading, ddtrace_threading):
@@ -23,42 +21,34 @@ cpdef get_thread_name(thread_id):
         # we fail, it'll just be an anonymous thread because it's either
         # starting or dying.
         try:
-            return threading_mod._active[thread_id].name
+            return threading_mod._active[thread_id]
         except KeyError:
             try:
-                return threading_mod._limbo[thread_id].name
+                return threading_mod._limbo[thread_id]
             except KeyError:
                 pass
 
     return None
 
 
-cpdef get_thread_native_id(thread_id):
-    # Do not force-load the threading module if it's not already loaded
-    if "threading" not in sys.modules:
-        return None
+cpdef get_thread_name(thread_id):
+    thread = get_thread_by_id(thread_id)
+    return thread.name if thread is not None else None
 
-    import threading
 
-    try:
-        thread_obj = threading._active[thread_id]
-    except KeyError:
-        try:
-            thread_obj = ddtrace_threading._active[thread_id]
-        except KeyError:
-            # This should not happen, unless somebody started a thread without
-            # using the `threading` module.
-            # In that case, well… just use the thread_id as native_id 🤞
-            return thread_id
+cpdef get_thread_native_id(thread_id):
+    thread = get_thread_by_id(thread_id)
+    if thread is None:
+        return thread_id
 
     try:
         # We prioritize using native ids since we expect them to be surely unique for a program. This is less true
         # for hashes since they are relative to the memory address which can easily be the same across different
         # objects.
-        return thread_obj.native_id
+        return thread.native_id
     except AttributeError:
         # Python < 3.8
-        return hash(thread_obj)
+        return hash(thread)
 
 
 # cython does not play well with mypy

ddtrace/profiling/collector/_task.pyx

@@ -1,7 +1,7 @@
 import sys
+from types import ModuleType
 import weakref
 
-from ddtrace.internal import compat
 from ddtrace.vendor.wrapt.importer import when_imported
 
 from .. import _asyncio
@@ -27,7 +27,7 @@ def install_greenlet_tracer(gevent):
 
     class DDGreenletTracer(object):
         def __init__(self, gevent):
-            # type: (...) -> None
+            # type: (ModuleType) -> None
             self.gevent = gevent
 
             self.previous_trace_function = settrace(self)
@@ -112,16 +112,22 @@ cpdef list_tasks(thread_id):
 
     tasks = []
 
-    # We consider all Thread objects to be greenlet
-    # This should be true as nobody could use a half-monkey-patched gevent
     if _gevent_tracer is not None:
-        tasks.extend([
-            (greenlet_id,
-             _threading.get_thread_name(greenlet_id),
-             greenlet.gr_frame)
-            for greenlet_id, greenlet in list(compat.iteritems(_gevent_tracer.greenlets))
-            if not greenlet.dead
-        ])
+        if type(_threading.get_thread_by_id(thread_id)).__name__.endswith("_MainThread"):
+            # Under normal circumstances, the Hub is running in the main thread.
+            # Python will only ever have a single instance of a _MainThread
+            # class, so if we find it we attribute all the greenlets to it.
+            tasks.extend(
+                [
+                    (
+                        greenlet_id,
+                        _threading.get_thread_name(greenlet_id),
+                        greenlet.gr_frame
+                    )
+                    for greenlet_id, greenlet in dict(_gevent_tracer.greenlets).items()
+                    if not greenlet.dead
+                ]
+            )
 
     policy = _asyncio.get_event_loop_policy()
     if isinstance(policy, _asyncio.DdtraceProfilerEventLoopPolicy):

ddtrace/profiling/collector/stack.pyx

@@ -2,12 +2,12 @@
 from __future__ import absolute_import
 
 import sys
-import threading as ddtrace_threading  # this is ddtrace's internal copy of the module, not the application's copy
 import typing
 
 import attr
 import six
 
+from ddtrace import _threading as ddtrace_threading
 from ddtrace import context
 from ddtrace import span as ddspan
 from ddtrace.internal import compat
@@ -309,7 +309,11 @@ cdef stack_collect(ignore_profiler, thread_time, max_nframes, interval, wall_tim
     exc_events = []
 
     for thread_id, thread_native_id, thread_name, thread_pyframes, exception, span, cpu_time in running_threads:
-        thread_task_id, thread_task_name, thread_task_frame = _task.get_task(thread_id)
+        if thread_name is None:
+            # A Python thread with no name is likely still initialising so we
+            # ignore it to avoid reporting potentially misleading data.
+            # Effectively we would be discarding a negligible number of samples.
+            continue
 
         tasks = _task.list_tasks(thread_id)
 
@@ -320,9 +324,6 @@ cdef stack_collect(ignore_profiler, thread_time, max_nframes, interval, wall_tim
             if task_pyframes is None:
                 continue
 
-            if task_id in thread_id_ignore_list:
-                continue
-
             frames, nframes = _traceback.pyframe_to_frames(task_pyframes, max_nframes)
             if nframes:
                 stack_events.append(
@@ -344,8 +345,8 @@ cdef stack_collect(ignore_profiler, thread_time, max_nframes, interval, wall_tim
                 thread_id=thread_id,
                 thread_native_id=thread_native_id,
                 thread_name=thread_name,
-                task_id=thread_task_id,
-                task_name=thread_task_name,
+                task_id=None,
+                task_name=None,
                 nframes=nframes,
                 frames=frames,
                 wall_time_ns=wall_time,
@@ -362,8 +363,8 @@ cdef stack_collect(ignore_profiler, thread_time, max_nframes, interval, wall_tim
                 thread_id=thread_id,
                 thread_name=thread_name,
                 thread_native_id=thread_native_id,
-                task_id=thread_task_id,
-                task_name=thread_task_name,
+                task_id=None,
+                task_name=None,
                 nframes=nframes,
                 frames=frames,
                 sampling_period=int(interval * 1e9),

releasenotes/notes/fix-profiler-gevent-tasks-76ed862362210cb7.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    profiling: Corrects accounting of wall and CPU time for gevent tasks within
+    the main Python thread.

tests/profiling/collector/test_stack.py

@@ -215,15 +215,16 @@ def _dofib():
         for t in threads:
             t.join()
 
+    expected_task_ids = {thread.ident for thread in threads}
     for event in r.events[stack_event.StackSampleEvent]:
-        if event.thread_name is None and event.task_id in {thread.ident for thread in threads}:
+        if event.task_id in expected_task_ids:
             assert event.task_name.startswith("TestThread ")
             # This test is not uber-reliable as it has timing issue, therefore
             # if we find one of our TestThread with the correct info, we're
             # happy enough to stop here.
             break
     else:
-        pytest.fail("No gevent thread found")
+        pytest.fail("No gevent threads found")
 
 
 def test_max_time_usage():

tests/profiling/simple_program_gevent.py

@@ -1,15 +1,17 @@
 from gevent import monkey
 
+# Import from ddtrace before monkey patching to ensure that we grab all the
+# necessary references to the unpatched modules.
+from ddtrace.profiling import bootstrap
+import ddtrace.profiling.auto  # noqa
+from ddtrace.profiling.collector import stack_event
+
 
 monkey.patch_all()
 
 import threading
 import time
 
-from ddtrace.profiling import bootstrap
-import ddtrace.profiling.auto
-from ddtrace.profiling.collector import stack_event
-
 
 def fibonacci(n):
     if n == 0:
@@ -20,7 +22,6 @@ def fibonacci(n):
         return fibonacci(n - 1) + fibonacci(n - 2)
 
 
-# When not using our special PeriodicThread based on real threads, there's 0 event captured.
 i = 1
 for _ in range(50):
     if len(bootstrap.profiler._profiler._recorder.events[stack_event.StackSampleEvent]) >= 10: