数据狗跨度在 python 线程池中丢失

Question

我有一个在线程池中运行的函数，但是当我在线程池之外运行它时，它只会显示在 Datadog 跟踪 UI 中。在下面的屏幕截图中，您可以看到它显示在 sync_work 中，但不在 async_work 中。

这是我的代码，包含在一个名为 ddtrace_threadpool_example.py 的脚本中：

from concurrent.futures import ThreadPoolExecutor
from ddtrace import tracer
from random import random
import time


def perform_work(input_value):
    with tracer.trace('do_something') as _:
        seconds = random()
        time.sleep(seconds)
        return input_value**2


def sync_work(input_values):
    with tracer.trace('sync_work') as _:
        results = []
        for input_value in input_values:
            result = perform_work(input_value=input_value)
            results.append(result)
        return results


def async_work(input_values):
    with tracer.trace('async_work') as _:
        thread_pool = ThreadPoolExecutor(max_workers=10)
        futures = thread_pool.map(
            lambda input_value:
            perform_work(input_value=input_value),
            input_values
        )
        results = list(futures)
        return results


@tracer.wrap(service='ddtrace-example')
def start_work():
    input_values = list(range(15))
    sync_results = sync_work(input_values=input_values)
    print(sync_results)
    async_results = async_work(input_values=input_values)
    print(async_results)


if __name__ == '__main__':
    start_work()

我这样运行脚本：python ddtrace_threadpool_example.py。我使用的是 Python 3.7，pip freeze 显示 ddtrace==0.29.0。

Answer 1

在与 Datadog 支持人员交谈后，这似乎是一个已知问题。

感谢您在我们调查此问题时的耐心等待。我们目前正在与 PoolExecutors 一起对此进行调查，并将提供更新。现在看起来异步调用中的那些子跨度丢失了上下文，因此它们似乎断开连接。

目前的解决方法是传入父级的上下文。在调用线程池执行器之前添加这一行。

current_context = tracer.get_call_context()

然后将该上下文传递给在线程池中运行的函数：

perform_work(
    input_value=input_value,
    parent_context=current_context
)

并使用它在函数内部创建一个跨度，如下所示：

span = tracer.start_span('do_something', child_of=parent_context)
seconds = random()
time.sleep(seconds)
span.finish()

完整的例子如下：

from concurrent.futures import ThreadPoolExecutor
from ddtrace import tracer
from random import random
import time


def perform_work(input_value, parent_context=None):
    span = tracer.start_span('do_something', child_of=parent_context)
    seconds = random()
    time.sleep(seconds)
    span.finish()
    return input_value ** 2


def sync_work(input_values):
    with tracer.trace('sync_work') as _:
        results = []
        for input_value in input_values:
            result = perform_work(input_value=input_value)
            results.append(result)
        return results


def async_work(input_values):
    with tracer.trace('async_work') as _:
        current_context = tracer.get_call_context()
        thread_pool = ThreadPoolExecutor(max_workers=10)
        futures = thread_pool.map(
            lambda input_value:
            perform_work(
                input_value=input_value,
                parent_context=current_context
            ),
            input_values
        )
        results = list(futures)
        return results


@tracer.wrap(service='ddtrace-example')
def start_work():
    input_values = list(range(15))
    sync_results = sync_work(input_values=input_values)
    print(sync_results)
    async_results = async_work(input_values=input_values)
    print(async_results)


if __name__ == '__main__':
    start_work()

这将产生如下所示的结果：