Потеря данных в пуле потоков python

Question

У меня есть функция, которая запускается в пуле потоков, но она отображается только в пользовательском интерфейсе трассировки Datadog, когда я запускаю ее вне пула потоков. На скриншоте ниже видно, что он отображается в 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 и будем связываться с обновлениями. Прямо сейчас это выглядит так, что эти дочерние промежутки внутри контекста asyn c call lost, поэтому они кажутся отключенными.

Обходной путь на данный момент - пройти в родительском контексте. Добавьте эту строку непосредственно перед вызовом исполнителя пула потоков.

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()

Это даст результат, который выглядит следующим образом: