Propagate spans to tasks

[crusaderky]

• Part of User-defined spans #7860
• Blocked by Annotations and spans leak between threads and asyncio tasks dask#10340.

Allow spans information to be accessed by tasks.
This is useful, e.g.

• for logging
• when the task itself wants to spawn more tasks, e.g.

def process_df(uri):
    with span("load"):
        df = dd.read_parquet(uri)
    with span("reduce"):
        out = ...(df)
    client = get_client()
    secede()
    return out.compute()

client = Client()
with span("foo"):
    foo = client.submit(process_df, "s3://mybucket/foo.parquet", key="foo")
with span("bar"):
    bar = client.submit(process_df, "s3://mybucket/bar.parquet", key="bar")
client.gather([foo, bar])

The above example will produce spans

• ("foo", "load")
• ("foo", "reduce")
• ("bar", "load")
• ("bar", "reduce")

[crusaderky]

CC @ntabris @hendrikmakait

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

       20 files  +       1         20 suites  +1   11h 44m 11s ⏱️ + 1h 14m 51s
  3 704 tests +       1    3 595 ✔️ ±       0     106 💤 ±  0  3 ❌ +1 
35 826 runs  +1 348  34 075 ✔️ +1 266  1 748 💤 +81  3 ❌ +1 

For more details on these failures, see this check.

Results for commit d51b536. ± Comparison against base commit 67e073f.

♻️ This comment has been updated with latest results.

[ntabris]

Is there a simple example of how to use this for logging?

[crusaderky]

Is there a simple example of how to use this for logging?

You may retrieve the current span with dask.get_annotations().get("span").
You can do so in the client code as well as from inside a task.

[hendrikmakait]

Thanks, @crusaderky! CI failures appear to be unrelated.