Hi all I m currently running a deployment with a flow like s

Question

Hi all I m currently running a deployment with a flow like sorry please follow the thread for the original message

Anna Geller · Answer

welcome back Sang wave could you move the code blocks to the thread

Sang Young Noh · Answer

Sure

Sang Young Noh · Answer

Hi all I m currently running a deployment with a flow like ``` flow def my flow ctx FlowRunContext = get run context seconds late = pendulum now diff ctx flow run expected start time abs=False in seconds 1 if seconds late gt 10 return Failed message=f Flow started seconds late seconds late my task ``` so that if a run is past a certain time automatically return a failure with a message However when the cli when I try to execute a simplified example ``` flow def my flow ctx FlowRunContext = get run context return Failed message=f Flow started seconds late seconds late my task ``` So the second one should automatically fail this one seems to run with a successful flow without failing Is this meant to be happening So I get something like ```Loading flow from deployed location Running flow 14 04 04 075 | INFO | prefect engine Created flow run taupe coua for flow ng 14 04 04 076 | INFO | Flow run taupe coua Using task runner ConcurrentTaskRunner 14 04 04 355 | INFO | Flow run taupe coua Flow can run still current 14 04 04 500 | INFO | Flow run taupe coua Finished in state Completed ``` I was expecting something like Failed with some error messages

Anna Geller · Answer

basically the main message should be a problem description without the code thanks so much pray

Sang Young Noh · Answer

Sorry yeah haha

Sang Young Noh · Answer

Yeah so I m currently trying to design a controlled failure for a flow but running ```prefect deployment execute ``` for this particular flow doesnt seem to reproduce the Failure

Anna Geller · Answer

execute is equivalent to `python yourflow py`

Anna Geller · Answer

can you try to describe the problem from a business perspective in your main message I didn t fully understand yet what you are trying to do

Anna Geller · Answer

and for the code any chance you could make it to a full flow I could reproduce

Sang Young Noh · Answer

Sure I will to write

Anna Geller · Answer

awesome

Sang Young Noh · Answer

```def late polling ctx maxlate=datetime timedelta hours=12 delibrately kill the flow if the we have gone past the maxlate threshold date logger = get run logger return Cancelled message=f Flow is too late it went past the max late threshold if ctx flow run expected start time + maxlate lt datetime datetime now tz=pytz utc return Cancelled message=f Flow is too late it went past the max late threshold else <http logger info|logger info> Flow can run still current ``` ``` flow name = ng def ng daily polling flow to call the NG daily polling function ctx FlowRunContext = get run context late polling is the equivalent of the task handler we have made late polling ctx print This is a random flow running that should not print if we fail successfully ``` So I m trying to ensure that the flow doesnt work after a certain threshold of time when deployed which in the case is after 12 hours or so I was planning on using a conditional failure return and test it with a random deployment but I figured that if I removed the time conditional 12 hours and just returned Failure state immediately it would at least confirm the logic works out As you see I couldn t quite produce the Failure state

Anna Geller · Answer

I see so this is your solution to the problem of Run it real time all the time in a while loop fashion but only between 8 and 22

Sang Young Noh · Answer

yes

Sang Young Noh · Answer

Well run it in a schedule but if we have a deployment that is out of bounds in terms of time 12 hours then automatically fail

Sang Young Noh · Answer

But again Im literally failing at failing this haha

Anna Geller · Answer

let me think about it I ll get back to you later today or tomorrow thanks for sharing your use case

Sang Young Noh · Answer

Sure no worries

Kevin Kho · Answer

If you want a flow to run every hour between 8 and 22 can t you use a CronSchedule like this ```0 8 22 ``` this is for every hour but you can change it to the frequency you need

Sang Young Noh · Answer

No its no quite that Ive currently got a deployment schedule using rrules but after its deployed I wish to make it fail after a certain number of hours in this case 12 has past after it has been deployed The deploymentschedule at the moment is ```FlowSchedule = rrule freq=MINUTELY interval=5 dtstart=dt 2020 1 1 8 31 0 byhour=range 8 9 byminute=range 30 52 return DeploymentSpec flow=ng daily polling name=f test 2 schedule=RRuleSchedule from rrule FlowSchedule tags=deploy tags parameters=parameters flow runner=SubprocessFlowRunner ```

Sang Young Noh · Answer

I may not have explained the problem properly so my apologies on my side

Kevin Kho · Answer

Ah you want a Deployment to only last for 12 hours Or you really want to fail flows

Sang Young Noh · Answer

So I wish to bake in every flow in a deployment that if for some reason if it is delayed beyond 12 hours then even if it runs fail it

Kevin Kho · Answer

Oh delayed I understand now Then I think your code above is how I d do the same Why does it fail If it s because of the Cancellation ``` return Cancelled message=f Flow is too late it went past the max late threshold ``` Instead you can do something like ``` task def fail task raise TimeoutError flow def myflow fail task result will be raised and end execution``` maybe

Sang Young Noh · Answer

So after the state of cancelled has been returned if I call upin the fail task result the flow should be cancelled So I am thinking ``` flow name = ng def ng daily polling flow to call the NG daily polling function ctx FlowRunContext = get run context late polling is the equivalent of the task handler we have made if late polling ctx fail task result print This is a random flow running that should not print if we fail successfully ``` something like this

Kevin Kho · Answer

I wouldn t rely on Cancellation I would just return True False I think from `late polling` And then fail my flow

Sang Young Noh · Answer

Hmm ok I will give it a try Thank you for the advice

Sang Young Noh · Answer

Just another question no rush on this I ve been running deployments of different flows with the same tags ```tags = run type 1 ``` which creates a work queue of the following ```WorkQueue id= 0bd02e1e 523a 4151 9d92 ea3d902e8a44 created= 27 seconds ago updated= 27 seconds ago filter=QueueFilter tags= run type 1 name= work queue type 1 ``` From my interpretation of it this workqueue should therefore be able to run both flows correct I did a preview of the work queue however and I only seem to getting the id of a single type of deployment So just wondering are the reployments and the tags only supportive of a single flow Or can it support multiple flows

Kevin Kho · Answer

It can support multiple Can you `inspect` your work queue through the CLI

Sang Young Noh · Answer

ok I ll have a look

Kevin Kho · Answer

I suspect you specified a deployment in there

Sang Young Noh · Answer

Running ```prefect work queue ls``` I get ```0bd02e1e 523a 4151 9d92 ea3d902e8a44 work queue type 1 None 7b09b84f ce57 42f2 90c1 a42b1dc16999 newworkqueue None ``` work queue type 1 is the one with tags From that I do ```prefect work queue inspect 0bd02e1e 523a 4151 9d92 ea3d902e8a44 ``` and I get ```WorkQueue id= 0bd02e1e 523a 4151 9d92 ea3d902e8a44 created= 6 hours ago updated= 6 hours ago filter=QueueFilter tags= run type 1 name= work queue type 1 ```

Sang Young Noh · Answer

I think I did something wrong but its good to hear that work queues does indeed support multiple deployments

Kevin Kho · Answer

Yeah that looks good and yes it should support