Невозможно использовать диапазон OpenTelemetry с генераторами, объединенными через aiostream.stream.merge. ⇐ Python

[Anonymous]

Мои диапазоны opentelemetry теряют своих родителей, когда я создаю их в асинхронном генераторе, который затем объединяю с другим асинхронным генератором.
Я считаю, что проблема в том, что диапазон, созданный в потокеstream_fn, теряется, поскольку объединенный поток переключается между контекстами в цикле событий во время ожидания.
Теперь я знаю, что могу вручную восстановить контекст после ожидания в этом сценарии, но в моем реальном сценарии восстановление контекст вручную при каждом ожидании будет очень неуклюжим и подвержен ошибкам.
как правильно передать контекст при объединении потоков?
Вот минимальное воспроизведение в виде автономного pytest
from collections import defaultdict
from aiostream import stream
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.trace import get_current_span

async def test_recurse_streaming_context():
trace.set_tracer_provider(TracerProvider())

async def stream_fn(title):
with tracer.start_as_current_span(title):
context = get_current_span().get_span_context()
yield (title, format(context.span_id, "016x"))
await asyncio.sleep(0)
context2 = get_current_span().get_span_context()
yield (title, format(context2.span_id, "016x"))

acc = defaultdict(list)
tracer = trace.get_tracer("test_tracer")
with tracer.start_as_current_span("test span"):
async for e in stream.merge(*[stream_fn(f"child_{i}") for i in range(2)]):
acc[e[0]].append(e[1])

assert acc["child_0"][0] == acc["child_0"][1]
# AssertionError: assert 'caa5694e43a514db' == '57adf8fb9e10a596'```


Подробнее здесь: https://stackoverflow.com/questions/781 ... stream-mer