hey I attempted to add an on fail slack alert with a decorat

Question

hey I attempted to add an on fail slack alert with a decorator like so ```def Flow flow args flow kwargs basically the same as prefect flow but adds Slack webhook def decorator func wraps func def wrapper args kwargs try return func args kwargs except Exception as e slack webhook block = SlackWebhook load alerts prefect msg = f Your job func name failed with exception e n msg += f See the flow run at <https prefect myinstance net flow runs flow run prt flow run id > n msg += f Tags prt flow run tags n msg += f Scheduled start time = prt flow run scheduled start time slack webhook block notify msg raise e return flow flow args flow kwargs wrapper return decorator``` it works but it pings for each retry of the flow any tips to make my flow ping only once the whole thing fails not just a retry

Nate · Answer

hi < Andy Dienes> seems like a great use case for the `on failure` hook ```from prefect import flow from prefect blocks core import Block from prefect server schemas core import Flow FlowRun from prefect server schemas states import State from prefect settings import PREFECT API URL def notify slack flow Flow flow run FlowRun state State slack webhook block = Block load slack webhook test slack webhook block notify f Your job flow run name entered state name with message n n gt state message n n f See lt https PREFECT API URL value flow runs flow run flow run id |the flow run in the UI gt n n f Tags flow run tags n n f Scheduled start time = flow run expected start time n flow on failure= notify slack retries=1 def noisy flow raise ValueError oops if name == main noisy flow ```

Nate · Answer

image png

Nate · Answer

```10 18 10 817 | INFO | Flow run radical oarfish Received non final state AwaitingRetry when proposing final state Failed and will attempt to run again ``` the flow will enter `AwaitingRetry` and then try again before entering `Failed` so it won t run the hook until it enters `Failed` at the end

Andy Dienes · Answer

looks great thanks I ll try it

Andy Dienes · Answer

hey Nate I tried this but it doesn t actually seem to work as in I know my slack webhook block works since my decorator pings it but with this formulation the `on failure` does not ping slack is there a certain version I need to be on I think I m currently on 2 10 4

Nate · Answer

i believe i was on 2 10 7 when i put together and tested the example i shared above but i dont think anything changed between 2 10 4 and now just tried again on 2 10 4 and it works for me

Nate · Answer

can you share your trace maybe your webhook block is not loading correctly from the hook

Andy Dienes · Answer

yeah I see the issue now I changed your `expected start time` to `flow run scheduled start time` and it doesn t have that field I suppose this is a little confusing to me since I had previously been using `prefect runtime flow run scheduled start time` without issue

Andy Dienes · Answer

does `prefect runtime flow run` not return a `FlowRun` type

Nate · Answer

hmm let me check on that

Nate · Answer

no i dont believe it does it looks like it only surfaces select flow run attrs from the flow run context

Andy Dienes · Answer

I see the reason I prefer it is that I have these jobs set on a schedule e g `0 9 ` and I want to see the timestamp it was supposed to run be logged if I use `expected start time` then it gives me whatever random timestamp at when the worker actually started which is usually a few millis off of the correct timestamp but in the worst case if my pool is unhealthy and the job is late then it can be many hours off of the logical time I wanted

Andy Dienes · Answer

also even with fixing that it still notifies for each failure I really want it to do all the retries first and then notify at the end only once this is my decorator call ` flow on failure= notify slack retries=1 `

Andy Dienes · Answer

in fact it pings me 9x sweat smile

Andy Dienes · Answer

the worker logs include this ``` ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File usr local lib python3 11 site packages prefect utilities asyncutils py line 91 in run sync in worker thread return await anyio to thread run sync ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File usr local lib python3 11 site packages anyio to thread py line 31 in run sync return await get asynclib run sync in worker thread ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File usr local lib python3 11 site packages anyio backends asyncio py line 937 in run sync in worker thread return await future ^^^^^^^^^^^^ File usr local lib python3 11 site packages anyio backends asyncio py line 867 in run result = context run func args ^^^^^^^^^^^^^^^^^^^^^^^^ File usr local lib python3 11 site packages prefect kubernetes worker py line 703 in watch job for event in watch stream File usr local lib python3 11 site packages kubernetes watch watch py line 165 in stream for line in iter resp lines resp File usr local lib python3 11 site packages kubernetes watch watch py line 56 in iter resp lines for seg in resp stream amt=None decode content=False File usr local lib python3 11 site packages urllib3 response py line 624 in stream for line in self read chunked amt decode content=decode content File usr local lib python3 11 site packages urllib3 response py line 816 in read chunked with self error catcher File usr local lib python3 11 contextlib py line 155 in exit self gen throw typ value traceback File usr local lib python3 11 site packages urllib3 response py line 461 in error catcher raise ProtocolError Connection broken r e e urllib3 exceptions ProtocolError Connection broken InvalidChunkLength got length b 0 bytes read InvalidChunkLength got length b 0 bytes read ``` maybe that is causing more retries than intended

Nate · Answer

hmm i will try to reproduce this