Is there a way to cancel the flows from cli (prefect 1)

Hey folks, me again. 🙂 I'm trying to get a grip as to how Prefect behaves when you have somewhat fine-grained tasks and try to enforce a concurrency limit on them. I'm running it on a GKE cluster with autopilot. My understanding is that tag-based concurrency limits are enforced clusterwise, meaning there must be some central mechanism for coordination, is that assumption correct? Then, I'm trying the following flow, which is the kind of thing I'll have to run when doing external API calls to enrich thousands of data items:

import time

from prefect import flow, task

from lib.prefect import get_run_logger


@task
def get_inputs():
    return range(1, 600)


@task(tags=['sequential-scraper'])
def process_input(item):
    logger = get_run_logger()
    <http://logger.info|logger.info>(f'Processing item {item}')
    time.sleep(1)
    <http://logger.info|logger.info>(f'Processed item {item}')


@flow(description='Example of a very simple, API scraping-like Pipeline. ' 
                  'Feel free to run it as it does not do anything harmful.')
def sample_pipeline():
    new_tokens = get_inputs()
    process_input.map(new_tokens)

And what I see essentially is my task runner (Job) taking a very long time until it gets wiped out without running any tasks at all, and the UI (Prefect cloud) leaves the flow in an eternal "running" state. Enabling debugging shows a lot of:

16:18:26.175 | DEBUG   | prefect.engine - Received wait instruction for 30s: Concurrency limit for the sequential-scraper tag has been reached
16:18:26.868 | DEBUG   | prefect.engine - Received wait instruction for 30s: Concurrency limit for the sequential-scraper tag has been reached
16:18:27.375 | DEBUG   | prefect.engine - Received wait instruction for 30s: Concurrency limit for the sequential-scraper tag has been reached
16:18:28.344 | DEBUG   | prefect.engine - Received wait instruction for 30s: Concurrency limit for the sequential-scraper tag has been reached

Am I misusing this in any way?

Hi Team. I frequently see this error in our flows. It occurs randomly across flows. I am not sure where to investigation from. Can you provide any insights into the error?

Crash detected! Request to <http://prefect-url:4200/api/task_runs/e26b6472-281f-4af1-933c-82516953f4a7/set_state> failed.

This message was deleted.

I’m seeing following exception irregularly on our prefect agent ec2

Nov 04 18:03:30 ip-172-31-30-84 prefect[242862]: 18:03:30.621 | ERROR   | prefect.agent - Invalid input ConnectionInputs.RECV_PING in state ConnectionState.CLOSED
Nov 04 18:03:30 ip-172-31-30-84 prefect[242862]: Traceback (most recent call last):
Nov 04 18:03:30 ip-172-31-30-84 prefect[242862]:   File "/home/prefect/insights-prefect-flows/venv/lib/python3.10/site-packages/h2/connection.py", line 224, in process_input
Nov 04 18:03:30 ip-172-31-30-84 prefect[242862]:     func, target_state = self._transitions[(self.state, input_)]
Nov 04 18:03:30 ip-172-31-30-84 prefect[242862]: KeyError: (<ConnectionState.CLOSED: 3>, <ConnectionInputs.RECV_PING: 14>)

Full trace in attachment This started to happen as soon as we upgraded from prefect 2.4.0 to 2.6.5 We are using Prefect Cloud. Is this a known issue? 😕

Hi everyone, I am trying to test the new run_deployment() that was released with 2.5 version, I have upgraded prefect in my local to 2.6.6 however I am getting this error --> ImportError: cannot import name 'run_deployment' from 'prefect.client.utilities' any inputs?

Hey folks! I'm working in Presto 1.0 trying to sort out how to debug this error:

Version group 92f0a1ae-7793-41bd-a662-d4cabbe103bf has no unarchived flows

This same error crops up when I use the Flow ID itself. I'm trying to do this with CURL just to play with the UI and build some code out as I can't just use the python lib due to some restrictions. Can someone check out the code in the thread and let me know if you have any thoughts here?

Hi guys, What's the proper way of running a one time only schedule flow in Orion? I can't find the DatesClock class anymore in Orion. In Prefect 1.X i was using something like this to start the flow 10 seconds after the registration:

def get_schedule(schedule: str):
    return Schedule(clocks=[DatesClock([pendulum.now().add(seconds=10)])]) if schedule == "one time" else CronSchedule(cron=schedule)

Thanks!

Hello! New to prefect. I am using a free version of prefect cloud. However one of my work queues is “unhealthy” and scheduled run is “late”. Wondering how to remedy, and what an “unhealthy” queue means.

This message was deleted.

any help on deleting a block type that I registered?

An exception occurred.
➜  flows prefect blocks type delete listblock
Cannot delete Block Type 'listblock'!

I can verify the block type slug is correct

Prefect 2.6.6, local. Any idea why my flows starting at the same time, although I have put a different schedule(start time)? Cheers

Hi all -- I am reading the docs for setting up deployments and hoping to set one up for my demo flow. The crux of my problem is that I have a set up similar to this:

def main(
    sbid,
    my_args,
    cluster,
):

    dask_runner = get_dask_runner(cluster)

    # Define flow
    @flow(
        name=f"Processing holography -- {sbid}",
        task_runner=dask_runner,
    )
    def my_flow():
        logger = get_run_logger()

        task_super_awesome_worker()

    my_flow()

This is a simplified example - but in short I am creating my flow function encapsulated in another function. I am going this as some attributed of the flow's task runner need to be defined at runtime depending on the SLURM cluster it is being executed on, account running the flow, compute resources for the slurm job etc (specified via the

dask_jobqeue.SLURMCluster

module/class). On the CLI my attempts to set the entrypoint

prefect deployment build /path/to/my/flow_script.py:my_flow

results in an error that amounts to

my_flow

not being in

flow_script.py

. If I try to replace

my_flow

with

main

, I get an error about

main

not being a

Flow

. In the docs I do not see an example of how to do this. So, I am wondering how does one do something like this?

Hi (Prefect migration question) In Prefect 1.0 I could see all the agents and could setup an “Automation” that monitors agent status and sends an alert if an agent stops working. How would you do those two things in Prefect 2.0? What is the equivalent to the triggers functionality from Prefect 1.0 https://docs-v1.prefect.io/api/latest/triggers.html#functions (is it the tasks

.submit()

,

.wait()

, `.result()`… ?) In Prefect 1.0 the “task” had “timeout” and “retry_delay”, so that I was able to limit the run time of a task, but wait some time before trying to run it again. (this is useful since sometimes a task get stack due to some cluster resource issue, and I want to let some time pass before rerunning it) (https://docs-v1.prefect.io/api/latest/core/task.html) How would you do such a thing with Prefect 2.0 ? (I see that t he flow decorator has those options, but not the task)

So DST caused some havoc in my flows today. After the time change, every flow has been running two times, once at it's normal time, and again one hour in the future. I set them using cron string in my yaml and adjusting the timezone to my local and thought based on documentation that it would auto adjust (pictured). So what's the plan of action here, is it something I can change in the UI, or do I have to re-deploy all of these flows?

Hello. We just upgraded our server/agents from 2.6.4 to -> 2.6.6. We are seeing resetting connections waiting for run_deployment. Thoughts? Thread for details

prefect 2.6.6 local server: I am using run_deployment() to run two flows i.e. Flow A and Flow B. second flow starts after first flow finishes. Whenever I triggers the flows from run_deployment() flow. the flow fails with an error although the Flow A completes without an error.

Hi all! I’m assessing prefect2 as a potential job orchestration solution. We have a mixture of python / non-python jobs packaged via Docker and we run on k8s. It seems like flows consisting of https://docs-v1.prefect.io/api/latest/tasks/kubernetes.html#runnamespacedjob best suits my needs, but is there a similar concept in prefect 2? Or is prefect2 not really the right tool to orchestrate arbitrary dockerized jobs?

Is there a way to run a flow locally and NOT have it show up on the UI ( 2.0 ) ?

What is the correct way of specifying a 'dummy' storage block for a deployment? My code that I have wrapped in prefect tasks correctly deals with absolute paths. Additionally, the data I am working with are on the order of 2TB, and are not really appropriately set up for the default local file system logic used when

--storage-block

is unset to copy. I thought I could set up a 'dummy' local file system block in the Orion UI (in my own managed server, not prefect cloud), however the

prefect deployment build

command says

'github', 's3',  'gcs', 'azure', 'smb'

are supported types. TL;DR - I need to set a

--storage-block

in my deployment, and I am reasonably certain in my situation I do not want to be copying anything to / from different file systems and blocks. I have a common underlying filesystem at the HPC center, and my data are pretty large -- large enough where I can not reasonably expect copying to and from the disk to be feasible

Hi , can someone tell me how to pass Service Account Info in the prefect_gcp block. I wanted to pass the values from the secrets block but prefect 2 doesn't seem to support json in secret block. What is the best way to get GCP BigQuery client from a dictionary. i.e a way to pass a dictionary from any of the blocks so that I can pass the Service Account Info as a parameter to the to obtain the big query client.

Hi everyone, i'm trying to use postgres database as prefect backend, I added the connection string and i stared the orion, all tables created at my db but missing fields, and got this error. can anyone help me?

Hi! When using the Github storage block to connect to a private repo (using Prefect version 2.6.6), I get a password related error (details in the first reply) when running a flow, what am I doing wrong?

Hi. I'm not missing anything specific from this, but wanted to make sure: Prefect 2 does not have an alternative for Prefect 1's

Flow(...).validate()

, right?

Hi, when running prefect deployment build with an infrastructure block I want to override the values of an array in the yaml, does anyone here know how I can do that? 🤔

Hi All, I’m working on moving away from using the Bitnami/Postgres subchart to an Azure DB for Postgres instance. After fixing the dialect name, I’m currently running into this error (will post in thread).

Hi Everyone, My team is getting k8 deployment jobs where the job log states the flow run status cannot transition to

RUNNING

Engine execution of flow run '148d81ea-3cfe-4db1-a0d4-3f3f17748fb0' aborted by orchestrator: This run cannot transition to the RUNNING state from the RUNNING state.

Will add more details to thread

Hello guys, ever since i updated prefect to 2.5+ i can not access the default blocks from Prefect UI. I dont know if that only occurs for me, but when i downgrade prefect to 2.3 the blocks seems to show up again. Any solutions for that?

Hi Everyone, I am trying to trigger prefect flow deployed on cloud whenever file is landed to S3 bucket using AWS Lambda function by referring this documentation from Chris. Lambda function is failing with

"errorMessage": "HTTP Error 404: Not Found"

I am passing API URL

<https://api.prefect.io>

. I understand that this documentation is not latest and I also tried URL

<https://api.prefect.cloud/>

and getting same error. Request your inputs regarding this error in Prefect version 2.0

Hi Prefect Folks - Is it possible to adjust the log level of a running flow? 🧵