nvm, found the issue, in addition of 'running', 'pending' flow run is also counted for concurrency

All, I want to delete 200+ flow run that are in pending state, I can filter them in UI. However, I am not able to 'select all' and delete all at once. I have to select one check box at a time. Any short cut ?

Need help deploying Prefect Orion and Prefect Agent to ECS Fargate Clusters - Prefect 2.7.3 =========================================================================== Hi, I am trying to deploy Prefect Orion and Prefect Agent to ECS Fargate clusters. I've had a successful deployment with Docker-Compose but somehow I've been running around in circles deploying it to Fargate. Here's what I am trying to do - 1. Prefect Orion server - running on a Fargate task 1 with Public IP 2. Prefect Agent 1 - running on Fargate task 2, looks up to the deployed Orion server 3. Prefect Agent 2 - running on Fargate task 3, looks up to the deployed Orion server Before anything moves further, I plan on upgrading future tasks to run as ECS tasks (still need to play around with that, but it's something for later) What I have done so far is - create an ECR image built from custom flows and tasks that we have put in, create - 1. ECS Cluster 2. ECS Service and task for Prefect Orion that uses latest Prefect image from Docker-Hub 3. ECS Service and task for Agent-1 that uses custom image from ECR repo built over Prefect's latest docker image 4. ECS Service and task for Agent-2 that uses custom image from ECR repo built over Prefect's latest docker image I've put in a AWS CloudMap to generate a internal network that allows Prefect's agents to look up for Prefect Orion server using internally resolved DNS. This is running into few issues - 1. Prefect Orion starts but throws an error - Can't connect to Orion API at <CloudMap resolved DNS>/api or <CloudMap resolved DNS>:4200/api or 0.0.0.0:4200/api. The task definition does contain Prefect API listed as an environment variable and PostgreSQL DB fetched from AWS SecretsManager. The web UI is running up but can't seem to connect to the DB or show anything unders /setings. (Attached sreenshot) 2. Prefect agents on both the tasks are not starting up and fail with this error - Resourceinitializationerror: unable to pull secrets or registry auth: execution resource retrieval failed: unable to retrieve secret from asm: service call has been retried 5 time(s): failed to fetch secret arnawssecretsmanager:_**_** from secrets manager: RequestCanceled: request context canceled caused by: context deadline exceeded. The same secret passed to Prefect Orion Fargate task seem to work just fine.

Hi all, pardon me for the naïve question. The concurrency limit for the queue in prefect 2 is for setting the concurrency of the tasks submitted in the flows or number of flow run executions per deployment?

Is this a good place to ask questions about the prefect-snowflake task? It uses

.fetchall()

which isn't usable for large datasets. Any other attempts we've done to use similar logic but using a cursor fails. What is it about the prefect execution environment that prevents us from using snowflake cursors?

Hey team! We’re looking at upgrading from Prefect1 to 2 in the near future. Currently our prefect services are deployed in house (using terraform and what not), but this allows us to quickly spin up and down new testing environments which use prefect 1. However, I notice in prefect 2, there are the new concepts of storage blocks, and if we want to continue to use s3 as a host for our flows, it seems like we have to login to a newly created environment and manually create the s3 storage block. Is there a way (through the API, env variables, other) we can set up an s3 bucket to allow us to keep the entire setup as infrastructure-as-code?

Hi all, what does this mean? "Local storage is not supported for Docker" Does Prefect store the docker images on the specified remote storage block? Why cant it store it on a local drive?

Hi eveyone,

Hello everyone,Iam not able to connect to prefect cloud via my PC although i have all access.Its showing SSL error wheen i have connected to my organization VPN.Its connecting fine when iam not connected VPN.Can anyone suggest how to access prefect cloud while connecting to VPN.

Hi, we had configured Slack notifications to be triggered on failure and success. However we're noticing failure messages arent getting reported. What could be the potential reason for this? We're running a self hosted prefect server

hi ! some of my flows have started crashing with

Flow run infrastructure exited with non-zero status code -1.

no extra logs nothing, it also leaves the tasks in running state. any idea why this happens ?

Sending SMS on task failure - Prefect 2.0 - Plivo I use Plivo as SMS service, not Twilio in Prefect 1.0 I used

'state_handlers': [o.p_state_handler]

which sent me an SMS alert on task failure. how can i achieve this kind of behavior in Prefect 2.0? thanks

Prefect 2 task timeout_seconds looking at https://docs.prefect.io/concepts/tasks/#timeouts it looks like this should be allowed:

@task(timeout_seconds=1)

but I get an error:

TypeError: task() got an unexpected keyword argument 'timeout_seconds'

What is the proper way to set a timeout time for tasks in Prefect 2 ?

Quick question: If a

task

is configured with

cache_result_in_memory=False

, can its result only be consumed via

state.result()

/

future.result()

? Calling such tasks directly, i.e..

result = my_task()

, seems to raise the

MissingResult

error.

Hi, I'm trying to understand how remote-storage should be used to distribute flows to a heterogeneous execution environment. I've drawn a couple of diagrams to clarify my understanding. The first diagram shows the creation of a deployment in the orion server via the cli. Config goes to orion, code goes to the s3 remote storage bucket. The agent subscribes to the work queue. So far so good. The second diagram shows the prefect agent receiving an run event from the queue and spawning a

myapp

docker container for execution.

Hello, Anybody had already any trouble with the Dask order of execution? (Local dask execution) My flow is executed right with ConcurrentTaskRunner, but DaskTaskRunner executes tasks, which should be executed only when their dependencies are executed. First image -concurrent graph (correct), second dask one (incorrect) I have created a relative issue

Hi, we are considering using Airtable for a DB. We would need users who can access through UI (read only access) and also programmatic access and because Airtable is so user-friendly it is on our list to be considered. Has anyone experience accessing programmatically Airtable via API? I heard the API is not very good and only supports HTTP. Have you worked with any other DB that would fulfil these requirements? user-friendly UI for read-only-user and accessible securely via API Also any suggested link about how to orchestrate the input/output from Airtable with Prefect? Thanks in advance!

Hello guys, I get the following error when trying to run a flow using the DaskTaskRunner using KubeCluster:

Crash detected! Execution was interrupted by an unexpected exception: AttributeError: 'KubeCluster' object has no attribute '__qualname__'

For more context, i'll add some code in the thread. Thanks!

Hey, guys I have problem with agent after succesffully runned flow (all 'tasks" run and finish but seting state fails:

For more information check: <https://httpstatuses.com/422>

151847.351 | INFO | prefect.agent - Completed submission of flow run '5adc8699-d9a5-4042-85d0-0b072a4873d0' <frozen runpy>128 RuntimeWarning: 'prefect.engine' found in sys.modules after import of package 'prefect', but prior to execution of 'prefect.engine'; this may result in unpredictable behaviour 151848.970 | INFO | Flow run 'fantastic-tamarin' - Hello from demo-flow! 151848.995 | ERROR | Flow run 'fantastic-tamarin' - Crash detected! Execution was interrupted by an unexpected exception: httpx.HTTPStatusError: Client error '422 Unprocessable Entity' for url 'http://prefect-server:4200/api/flow_runs/5adc8699-d9a5-4042-85d0-0b072a4873d0/set_state'

I’m getting a bunch of misc exceptions from Prefect during my Flow runs. Errors in thread (most of them are

sqlite3.OperationalError

). Anyone know what’s going on? Context: I’m using the

@flow

and

@task

decorators for my Python functions in my app hosted on Google Cloud Run. I’m not using Prefect Cloud. I’m on the following prefect version:

prefect==2.3.2

. It’s hard for me to repo because this occurs during my chron job. I haven’t been able to repo locally.

I've managed to get my flows to run in docker infrastructure, using a remote s3 storage block for deploying flows, which is pretty cool (to me :P). (earlier struggles) I'm curious how people deploying/declaring/installing the prefect dependencies of their flows though 🤔 . Is it correct that the flows themselves are downloaded at runtime, but the dependencies have to be included in the infrastructure image ahead of time? Say I'm trying to orchestrate a Java application. Does the Java application image also need to have Python, prefect and the prefect dependencies pre-installed?

Firstly, thank you all for building and maintaining Prefect, I’ve been using it in production for a while now and it works wonderfully. Similar to the previous question, currently I’m trying to orchestrate a non Python application using Prefect. The ideia is to use the Prefect to periodically run PHP scripts inside Docker containers, however we might also use Prefect for other scripts in the future, written in other languages. I’ve managed to successfully run non-python scripts inside Docker containers through Prefect, but the agents that run these containers does not seem to update the server with the flow run status. Please see this thread to understand what I did.

Prefect V1 question - I think my flow concurrency limit is being used up by deleted flow runs. Is there a way to get the flow ids that are running with a particular label?

Hello! I have a few flows that take dataframe as input and return a transformed dataframe. I would like to create an orchestrator flow that calls the flows using

run_deployment

. What would be the most efficient way of passing the dataframes around?

allow_failure

does not work as expected I am trying to see how can I continue with a workflow after a task failed all the allowed retries. Sample code:

from prefect import task, flow, get_run_logger, allow_failure
from time import sleep


@task(retries=2, timeout_seconds=3)
def test_1(o, test_retries):
    if test_retries:
        sleep(5)
    else:
        raise IOError('Raising some error')
    return True


@flow(name='test allow_failure')
def run_test_flow():
    result = test_1.submit(True, True)
    result = test_1.submit(result, False, wait_for=allow_failure(result))


if __name__ == "__main__":
    run_test_flow()

I expect this to run

test_1.submit

, fail twice, since the

sleep

is larger than the

timeout_seconds

, and then run the second

test_1.submit

But it does not appear to be working it does not run the second

test_1.submit

@User, @User do you have thoughts on this ?

Is it correct that in order to run tasks concurrently we need to us

submit

? Even if the flow is written in a way that there is no dependencies between tasks?

Hi everyone,How can i get cache storage key for file in prefect.Iam using default caching in prefect.I want to get a storage id for file

Hi! I'm trying to use Deployment() to deploy my flow to an orion server. I'm not able to set the client right. Looks like Deployment().apply() is calling get_client() which gets the api_url from profile.toml. i want to set it from my script itself. How do I go about this?

Help with Prefect

v2

(more details in thread!) 1. 🏗️ Structuring project: We're using v2 with a GCS Storage block, a KubernetesJob block and a Flow Deployment (all three defined in Python files). Because we're using GCS Storage, the flow files are not

pip installed

so we can only use relative imports in the flow like

from .utils import foo

, which can make local development and bugchecking a bit more error-prone. We will probably shift to just using the Docker images local files, but haven't yet experimented with this in Prefect v2. Are there any best-practises around this? (Project structure in thread) 2. 🏃‍♀️ Running flows: Prefect

v1

had a

prefect run ...

command in the CLI, which made it possible to run a local flow. Prefect`v2` doesn't seem to have this? We can't add a

if __name__ == "__main__"

to the Flow file, because it has relative imports so running it doesn't work. As a workaround I've added a

run.py

in a parent directory that imports and runs it, but this isn't very ergonomic. In either case if I want to pass parameters, I must create my own CLI for doing so... Have I missed something in the CLI, or is there a plan to add this? Or is there a better pattern I should be using instead?

hi everyone, just a quick question, how can i get the current flow run name programmatically using python during the run? couldn’t find the doc atm…