Hello! I'm running a Prefect Server on K8s and instantiating a Dask cluster to run my flows. It seems to be working well (the flow finishes successfully), but it is not storing the task logs on the UI. Is there any way to guarantee it is going to be there? When I've added a

sleep(60)

before the end of the task, the task logs appeared, but without it it does not work.

Are you using a dask kubernetes cluster?

Yes, I'm using the KubeCluster class. This is my flow:

with Flow(
    "Dask Kubernetes Flow",
    storage=storage,
    executor=DaskExecutor(
        cluster_class=lambda: KubeCluster(make_pod_spec(image=prefect.context.image)),
        adapt_kwargs={"minimum": 2, "maximum": 3},
    ),
    run_config=KubernetesRun(),
) as flow:

Did this work before? Because Dask logs really don’t show in the UI

Oh, I see. This is something I'm trying this week, so I cannot say if it has worked before. Is there any reason that adding a sleep() on my task made so that the log appeared on the UI?

No I don’t at the moment. I wonder if it’s related to this. (Though the issue is for local). Maybe you can chime in there?

Sure, thanks. One last question, is there an easy way that people normally use to aggregate the logs of the clusters directly from K8s? I thought of using the

get_logs

method of KubeCluster, but given that Prefect is creating the cluster and then deleting them after running, I won't be able to run the command

Could you do it in Python mid run? Like you could use a state handler?

I see. I'll explore this approach and try to make something work. Thanks a lot!

No I am clueless about that at this time. Is it consistent (adding 60 seconds works every time)? Could you have a server time and flow time that are out of sync?

It seems it doesn't have to be just 60secs. But I have to add it on all my tasks I want to log currently. I've added a 5secs sleep on each of them and I got a lot more logs onto the UI