Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Prefect Community

Hello Everyone,

We currently have a setup of Prefect Core Server running on our on-prem environment using Kubernetes. We are running our flows using a LocalDaskExecutor and we noticed that the logs for these flows do not show up on the portal. Only logs for flows that do not use a LocalDaskExecutor show logs. Is there a special configuration required for Dask to send the logs?

There might be an <https://discourse.prefect.io/t/why-are-the-dask-logs-not-being-shown-in-the-prefect-ui/75|issue> with sending logs from all workers when you use DaskExecutor.

But LocalDaskExecutor doesn't use Dask so this shouldn't be a problem - can you share your example flow that fails to send logs to your Server backend?

Hi Anna, Thanks

```class RunMeFirst(Task):
    def run(self):
        print("I'm running first!")


class PlusOneTask(Task):
    def run(self, x):
        print(x + 1)
        return x + 1


flow = Flow('My Imperative Flow')
plus_one = PlusOneTask()
run_me = RunMeFirst()
plus_one.log_stdout = True
run_me.log_stdout = True
flow.set_dependencies(
    task=plus_one,
    upstream_tasks=[run_me],
    keyword_tasks=dict(x=10))

flow.storage = get_jigsaw_api_storage(1, f"Test Flow")
flow.storage.build()
flow.run_config = get_jigsaw_k8_config_run()
flow.executor = LocalDaskExecutor(scheduler="processes", num_workers=2)
flow.register("Utilities")```

Note that when i remove the flow.executor line before the last line , the prints appear normally in the logs on the UI

Thanks for sharing your code. Could you try running a simple hello world flow using the functional API and task decorator? This way you can step by step test what is failing in your configuration

e.g. you may try:
```from prefect import task, Flow
from prefect.executors import LocalDaskExecutor


@task
def generate_random_numbers():
    return list(range(1, 200))


@task
def add_one(x):
    return x + 1


@task(log_stdout=True)
def print_results(res):
    print(res)


with Flow("mapping", executor=LocalDaskExecutor(), storage=get_jigsaw_api_storage(1, f"Test Flow")) as flow:
    numbers = generate_random_numbers()
    result = add_one.map(numbers)
    print_results(result)```
The issue you have is that you build the storage *before* you attach the executor. Since the executor information is retrieved from storage at runtime, your executor information won't be found. Switching the order may already help solve your issue:
```flow.executor = LocalDaskExecutor(scheduler="processes", num_workers=2)
flow.run_config = get_jigsaw_k8_config_run()
flow.storage = get_jigsaw_api_storage(1, f"Test Flow")
flow.storage.build()
flow.register("Utilities")```


Hi Anna, I tried to change the order as you suggested and I'm still getting the same :disappointed:

will try to investigate more to understand what is happening

but do you understand the issue? the problem is that you are attaching the executor too late

Yes I think I got the problem, I'm trying to see how I can change the flow a little bit so I can capture the logs correctly