@Marvin Hello, I am currently trying to use prefect (using helm and values.yaml to setup work-pool and workers). The issue is each flow I try to run crashes due to validation error for KubernetesWorkerJobConfiguration env (value is not a valid dict)

Hi again 😄 I just got the error

No heartbeat detected from the remote task; marking the run as failed

- this happend for tasks being run in parallel via the dask

KubeCluster

looking at our (internal) logs - it seems like all of them died simultaneously, 20+ minutes into their run there’s a total of 58 tasks being run, with 4 workers the first 4 tasks took approx. 1 hour to run and completed successfully, and then the next 4 that were run all seem to have failed simultaneously any idea / help?

Hi all, I'm looking at configuring logging for our prefect setup. We use [structlog](https://www.structlog.org) across our python services to format our log entries which are typically sent to cloudwatch log groups; these are then ingested by centralised logging (ELK stack). I'd like to know if it's possible to intercept all logs getting sent to Prefect cloud and duplicate them to stdout, applying formatting (we use JSON). Probably a complicated request; any help much appreciated!

I'm encountering a persistent issue with Dask serialization when updating my flows from Prefect

2.19.1

to Prefect

3.1.2

. Every run incurs the following error, which I've never seen before in 18 months of using Prefect 2:

Task run failed with exception: TypeError('Could not serialize object of type HighLevelGraph', '<ToPickle: HighLevelGraph with 1 layers.\n<dask.highlevelgraph.HighLevelGraph object at 0x7f4e3b547400>\n 0. 139974151916736\n>') - Retries are exhausted
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/distributed/protocol/pickle.py", line 63, in dumps
    result = pickle.dumps(x, **dump_kwargs)
  File "/usr/local/lib/python3.10/dist-packages/prefect/docker/__init__.py", line 20, in __getattr__
    raise ImportError(f"module {__name__!r} has no attribute {name!r}")
ImportError: module 'prefect.docker' has no attribute 'isnan'

<---Long series of stack traces--->

The above exception was the direct cause of the following exception:

<<---Long stack trace--->>

  File "/usr/local/lib/python3.10/dist-packages/distributed/protocol/serialize.py", line 392, in serialize
    raise TypeError(msg, str_x) from exc
TypeError: ('Could not serialize object of type HighLevelGraph', '<ToPickle: HighLevelGraph with 1 layers.\n<dask.highlevelgraph.HighLevelGraph object at 0x7f4e3b547400>\n 0. 139974151916736\n>')

All of my flows write to Zarr N-Dimensional data stores using Xarray's

to_zarr

method,.

to_zarr

leverages Dask under the hood to vastly speed up this operation. I don't use Dask Task Runners or any specialized Prefect Dask objects. Digging around I've found this conversation, which doesn't seem directly related https://linen.prefect.io/t/23198687/hello-everyone-i-m-facing-an-issue-using-prefect-with-a-dask. Any ideas Prefect team? I'll post a full stack trace in follow up

@Marvin I create deployment, but it's not showing up in the Prefect UI

I'm nearly there on getting my first successful flow run with an ECS work pool, but I keep getting a Crashed run with this error:

Flow run infrastructure exited with non-zero status code:
 Exited with non 0 code. (Error Code: 1)
This may be caused by attempting to run an image with a misspecified platform or architecture.

I've been following this guide (https://docs.prefect.io/3.0/deploy/infrastructure-examples/serverless) and the

prefect work-pool create

step went fine. What am I missing here?

Hi, I am trying to run a prefect flow using a trigger from a Webhook. I am trying to figure out a way to send the data I received in the body of the webhook to the flow as a parameter, but I can’t figure out a way to do it. I followed the example given here but instead of passing

{{ event }}

(which is basically just the name of the event, I tried passing in

{{ resource.event_data }}

. Please note that

resource.event_data

is a property I defined this property in my webhook template, it gets populated from the body of the webhook I received originally. So, what’s happening is that the flow gets executed but instead of receiving the actual

resource.event_data

it just received the raw string “{{ resource.event_data }}“. Is it even possible to pass on dynamic data from the original event to the deployment as a parameter?

I'm having trouble with crashing async tasks and I'm not sure how to troubleshoot it. I'm using this pattern within a flow to start the tasks concurrently:

coros = []
for item in item_list:
    coros.append(task1(item , return_state=True))
    coros.append(task2(item , return_state=True))
    coros.append(task3(item , return_state=True))
    coros.append(task4(item , return_state=True))

    states = await asyncio.gather(*coros)

For a small

item_list

, everything works fine. But if

item_list

approaches 100 items, almost all of the async tasks crash. I'm not seeing anything helpful in the logs to indicate why the tasks crash, so I'm wondering how to go about troubleshooting/fixing this async flow for large lists. Each crashed task shows this error message:

Crash detected! Execution was cancelled by the runtime environment.

How can I figure out why execution was canceled? Is there any other logging source that I should be looking at outside of the Prefect logs? Or do I need to increase my log level for this type of issue? Or do I need to calculate how many workers are needed in the work pool to handle large lists? Background info (in case it's useful): • Prefect version 2.16.2 • The compute instance is very large. It has 12 cores and 62G memory. • The compute and memory footprint of each task is small (each is calling an API and waiting for a response) • I'm using work pools and all tasks are being sent to the same work pool.

@Marvin I have pgbouncer in front of my postgres instance and I got the following error when starting the server.

asyncpg/connection.py", line 2027, in _do_execute
    result = await executor(stmt, timeout)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "asyncpg/protocol/protocol.pyx", line 206, in bind_execute
asyncpg.exceptions.InvalidSQLStatementNameError: prepared statement "__asyncpg_stmt_2__" does not exist
HINT:
NOTE: pgbouncer with pool_mode set to "transaction" or
"statement" does not support prepared statements properly.
You have two options:

* if you are using pgbouncer for connection pooling to a
  single server, switch to the connection pool functionality
  provided by asyncpg, it is a much better option for this
  purpose;

* if you have no option of avoiding the use of pgbouncer,
  then you can set statement_cache_size to 0 when creating
  the asyncpg connection object.

But I cannot disaplbe this and I need to use statement_cache_size in my URL BUT if I try to append this to URL connection ?statemente_cache_size=0 I got the following error:

asyncpg/connect_utils.py", line 731, in _parse_connect_arguments
  if var_val is None or isinstance(var_val, bool) or var_val < 0:
                                                     ^^^^^^^^^^^

How can I start Prefect server to work with a postgres server that has pgbouncer on front of it without using session connection?

Ok so I updated to the latest and I'm still getting this bug:

File "/home/dave/.cache/pypoetry/virtualenvs/rustmonster-GKhQY2Y4-py3.10/lib/python3.10/site-packages/prefect/client/orchestration.py", line 219, in get_client
    server.start()
  File "/home/dave/.cache/pypoetry/virtualenvs/rustmonster-GKhQY2Y4-py3.10/lib/python3.10/site-packages/prefect/server/api/server.py", line 818, in start
    raise RuntimeError(error_message)
RuntimeError: Timed out while attempting to connect to ephemeral Prefect API server.