I’m still running into a lot of issues with async flows — raised this issue 10 days ago specifically related to subflows - but having massively stripped down the amount I’m using prefect (basically just using it to run the

main

function - with everything else being plain async python) and i’m still getting random hangs — and when I kill the script, it seems (as best as I can tell), that the

KeyboardInterrupt

gets caught somewhere within the prefect codebase. what is curious is that when I dropped the top-level

@flow

altogether and just ran as a plain python script, it hasn’t stalled yet (will update if it does). is it possible that there’s something going on with longer-running (5min+) async flows that’s causing them to hang at some point? I thought maybe it was to do with hitting the logging rate limit but by removing all of the task decorators I’ve managed to limit the overall prefect calls significantly. I’m racking my brain at this point. anyone have any experience with long-running async programs? are there other factors that could be causing random hangs?

When querying the

api/work_queues/[ID]/status

endpoint, is it possible to modify the health_check_policy? Here is a typical response:

{
  "healthy": true,
  "late_runs_count": 0,
  "last_polled": "2023-05-09T17:08:54.625892+00:00",
  "health_check_policy": {
    "maximum_late_runs": 0,
    "maximum_seconds_since_last_polled": 60
  }
}

Hi Team! Is there a way we can get prefect logs to also pick up logs from a .txt file? What's essentially happening is that we are using Prefect to kick off another job, which logs to a .txt file, but we want prefect to pick up those logs in real time as the process executes. I understand there's a logging.yml, is there a parameter there that could help us?

Is there a good reason calling sub-flows and/or

run_deployment()

doesn't expose a

PrefectFuture()

interface similar to

Task.submit()

? For a flow like:

@flow
def f():
  a = subflow_foo.submit(1)  # .submit() doesn't exist on Flow or run_deployment().
  b = subflow_foo.submit(2)
  c = subflow_foo.submit(3)

  x = subflow_bar(a, b)
  y = subflow_bar(b, c)

  x.wait()
  y.wait()

I want to get the engine's dependency tracking to kick off downstream sub-flows as soon as their dependencies are satisfied. It feels like I could wrap run_deployment() in a future myself, but maybe I'm missing some non-obvious reason this wouldn't work?

I can't find much other than a thread from 2021 (https://github.com/PrefectHQ/prefect/issues/4789) that describes this problem, but is there some way within the snowflake block's methods to get the column names coming back with snowflake query results?

Can anyone help me out with how to manage dynamic dependencies in prefect flows? I'm trying to create a pipeline in which a process is triggered and a python script is executued. Everytime a python script is submitted, first its dependencies should be installed, and then the script should be executed. How can I handle dynamic pip packages in this context? Tried to run prefect by docker in docker but the overhead is too much both from networking PoV and also resource wise. Currently trying to go for python venvs + a way to manage pip dependencies.

Hello all, When deploying flows from a python using an RRule scheduler to schedule daily updates, I’m consistently seeing odd behavior where a flow gets duplicated many, many times, 1-3 seconds apart. See below screenshot — ~100 more flows for 3 days follow, when I expected to see only 3. Any ideas what might cause this? The RRule scheduler in question is

FREQ=DAILY;BYHOUR=10;BYMINUTE=21

. Even more curiously, sometimes the schedule will seem to work correctly immediately after running. Then I will check back some hours or a day later and the duplication will have happened.

any1 else had issues where they are running agent on aws(fargate) and the flow is in running state but there isn't a task running in was anymore? I don't realise it unless I check the flow logs and notice that the last log was probably an hour ago or so.

Hello, I've set up an Automation in Prefect Cloud to send a slack notification when a work queue becomes unhealthy. I've used the same block (Slack Incoming Webhook, from the prefect_slack pip package) I have stored to send other slack notifications within the flows, which works fine. However, when the automation is triggered it fails with the following error:

The notification block was invalid: KeyError(\"No class found for dispatch key 'slack-incoming-webhook' in registry for type 'Block'.\")

Is this a bug?

Hello, I have one doubt, If we use a

task

with

cache_key_fn

, then that where is that cache is stored in Prefect cloud? If I use self hosted version then where this cache is stored and retrieved?

Hi there, I’m playing with Prefect Python SDK. It seems the

task

decorator is causing some issues with

mypy

. 🧵

Hello there I try to use new concept of projects and to deploy with

prefect_docker

template to build a docker image and to push it on my private registry (on Gcloud) but it looks like it's not possible to use an other registry than docker. Any idea? Otherwise, what's the best practice to deploy a docker image flow on a private storage?

Hello, While deploying a flow using the prefect python

Deployment

class and I have been seeing the following logs.

08:02:49.965 | WARNING | prefect.server.services.cancellationcleanup - CancellationCleanup took 26.55879 seconds to run, which is longer than its loop interval of 20.0 seconds.

Is it safe to disable Cancellation cleanup service on my temporary Prefect server that I use only for deployment?

how can I add job configurations to my deployment object if I wanna deploy it via a custom worker?

The docs are confusing.

Hello guys! I am trying to execute a function on failure for a flow run but the arguments which admit are just flow, flow_run, state I want to know the task name which fails and also the message value but I don't know how access to it.

on_failure=[function_to_execute]

def function_to_execute(flow, flow_run, state):
   #get task name an error

I am using prefect 2.10.10 version

what's the recommended concurrency limit per work queue and per work pool when using the default task runner?

Did Prefect deactivate schedules in prefect v1? My prefect v1 account is supposed to be scuttled on July 15th, but I saw all of our schedules got turned off sometime after Thursday morning last week. Is this something we can expect to see happening again, or has this been logged as an issue? I'm relying on being able to use this until July 15th

Hi All! I am working on transition to prefect 2 with docker deployment and storage (via the new projects config files) and am running into this error from the worker at runtime:

09:00:08.037 | ERROR  | Flow run 'cobalt-ladybug' - Flow could not be retrieved from deployment.

Based on this recipe on github I saw that “with image-based infrastructure blocks…the image can serve as a fourth storage option…In practice, when creating a deployment with an image-based infrastructure block, you can omit the storage block and Prefect will assume your flow code is available in the image at runtime.” I assume that when i run `prefect deploy`in the CLI, my flows will be bundles up and stored in the docker container, but i guess that is not the case. Can someone provide guidance? I attached my deployment.yaml and prefect.yaml files here to help debug

Hi there, I’m confused about how to deployments with a custom image. I have a Prefect Worker running on my Kubernetes cluster. Here is the

deployment.yaml

for my project:

deployments:
- name: my-kubernetes
  version: "0.0.1"
  entrypoint: demo.py:run_demo_prefect
  infrastructure:
    type: kubernetes-job
    image: "docker-repository-url/image-name:tag"
  work_pool:
    name: default
    work_queue_name: default

The worker correctly runs the deployment in a new Kubernetes Job, yet using the default Prefect image instead of the one I have specified. What I’m missing?

Hey everyone! I'm facing an issue while using the

prefecthq/prefect:2.10.11-python3.10-conda

image. Whenever I include the ENTRYPOINT

["prefect", "agent", "start", "-q", "default"]

, I encounter the following error: 'Failed to deploy '<unknown> Dockerfile: Dockerfile': Status 400: {"message":"failed to create shim task: OCI runtime create failed: runc create failed: unable to start container process: exec: "prefect": executable file not found in $PATH: unknown"}'. Interestingly, I don't encounter this problem with the

prefecthq/prefect:2.10.11-python3.10

image. Any ideas on how to resolve this?

Does deleting a queue delete all the scheduled flows? I was looking through the api and I could delete all flows using the API but that requires knowing all the flow ids. I'm not sure there is a way to GET all flow ids. So i'm thinking of just deleting the queue. In the UI sometimes deleting the queue give a 'failed to delete queue' popup, so willing to do that via the API if all flows scheduled and late in there are deleted too upon deletion

Hello, I am running a flow in kubernetes (2.10.10) and for some reason, when the flow run fails, it gets automatically retried creating multiple jobs before the flow run is finally marked as Failed. What could be causing this behaviour? I did not set any retry logic. In 2.9.0 In 2.9, I get two k8s jobs, the first one fails and then a second run is executed but it finishes right away with message

Engine execution of flow run '2c1f694b-3ca7-45f4-8048-908838963b52' aborted by orchestrator: This run has already terminated.

Any ideas? thanks!

<Automations function bug report> We have set the notification about a flow run stays in 'running' in 3hours. but it seems that automation function cannot detect 'canceled' status. For example, the flow in this screenshot was canceled but '3hours notification' worked. Could you check this bug~?

Hello, I am wanting to use prefect blocks for example, AWS Credentials. Although I get to only temporary access key ID and secret key. Not sure as to how to use aws config profile to do the same. Any suggestions?

Has anyone seen this error in a k8s: `Job did not complete within timeout of Nones`…I have a task that makes concurrent requests to an API for large amounts of data. We’ve scaled our cluster CPU and memory so that shouldn’t be an issue; while the task was running for ~4 hours, the flow crashed while making API requests and the error I got is the one I mentioned.

Hello. I have a flow thats permanently in a PENDING state? How do I begin to debug this issue? Other flows in the same work queue is RUNNING and completing successfully.

I am running the Helm version on Kubernetes, I am not able to start anything as each API call returns 405, even from the UI and if i use the prefect client. just dont know what to do . here are the images i am using. kindly help. prefecthq/prefect:2.10.9-python3.10 server prefecthq/prefect:2.10.9-python3.10 agent prefecthq/prefect:2.10.9-python3.11-kubernetes worker

Hi everyone, I have a big flow with hundreds of smaller tasks that run in parallel. Currently, a flow run is processed in only one k8s pod. Is it possible to process those tasks in multiple pods at the same time?

I am running the Helm version on Kubernetes, I am not able to start anything as each API call returns 405, even from the UI and if i use the prefect client. just dont know what to do . here are the images i am using. kindly help. prefecthq/prefect:2.10.9-python3.10 server prefecthq/prefect:2.10.9-python3.10 agent prefecthq/prefect:2.10.9-python3.11-kubernetes worker error logs from the server component