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

Hiya, trying out the `DaskKubernetesEnvironment` and am running into a weird issue.

Using this code <https://docs.prefect.io/orchestration/execution/dask_k8s_environment.html#examples> with a couple of changes to use our flow registration system. We're running into this error:

```State Message:
Failed to load and execute Flow's environment: ValueError('Storage for flow run dask_demo is not of type Docker and environment has no `image` attribute in the metadata field.',)```
But looking at the details  for our flow, it looks like everythings correctly set?

Environment

Type
*DaskKubernetesEnvironment*

Storage
path

type
D*ocker*
secrets
[*]*
image_tag
c*18514566fcd9b68dcd5c44e9adea9ab0defe7d5*
image_name
p*refect*
registry_url
*.<http://dkr.ecr.eu-west-1.amazonaws.com/prefect|dkr.ecr.eu-west-1.amazonaws.com/prefect>*

stored_as_script
*false*

Hey <@U014490S4EB>,
Digging in a bit, I’m seeing that this error looks like it occurs from the Agent not being able retrieve the flow from storage because it cannot find the metadata attached at registration time. It may be difficult to diagnose without the code for your storage, did you include the `registry_url` and `image_name` for the flow’s storage upon registration?

Hi Kyle, thanks for having a look. I believe we are (my message before is from the "details" tab of the flow itself, just a little mangled from formatting) we use the same registration code to run other KubernetesTaskFlows without issue.

Hmm, this is a tough one. <https://github.com/PrefectHQ/prefect/blob/b18d8aabb591133655b3196480f622b0a822daca/src/prefect/utilities/agent.py#L36|This> is the only place in which that error occurs, so I believe that leaves our environment - which may not be able to pick up that container as defined by that yaml…?