Structured logs can help here, example from the documentation:
def myfunc(x):
start = time()
...
stop = time()
dask.distributed.get_worker().log_event("runtimes", {"start": start, "stop": stop})
futures = client.map(myfunc, range(10))
client.get_events("runtimes")