Hey everyone, Been banging my head on the keyboard for a bit on this one, and would be indebted for any guidance. I’m trying to create a flow that: 1. adds metadata to FAILED signals via a mapped task 2. applies a state handler to the flow that can look up metadata from the FAILED task states and alert the user with a final notification that includes detailed metadata from each failed task The strategy is based on the notification docs: https://docs.prefect.io/core/concepts/notifications.html#responding-to-state When I execute a basic flow run locally (without using server or Cloud), the script works as expected and I get a log message listing the failures (I’m testing with a logger, but the goal is to eventually alert via Slack). But when I run the same code against Server or Cloud, the individual task failures no longer seem accessible via the state handler. It seems like

new_state.result

inside the state handler is the problem locale. Locally, it is a dict populated with (<task>, <state>) items; on Server or Cloud,

new_state.result

appears to return an empty dict. I’ve created two basic flow scripts to demonstrate this behavior (one for a local manual run, the other for Server or Cloud) and will attach them below. Each contains a

NOTE

marking the divergent status of

new_state.result

. Would be indebted if someone could advise on what I’m doing wrong!

Looking at Prefect's code without testing, so I may be suggesting gibberish, I wonder if you could try:

raise signals.FAIL(e, result=(x, y))

If I'm reading the code right, this should set the result of the

Failed

state associated with the

FAIL

signal to

(x, y)

, so in the state handler

new_state.result

should have it.

@Amanda Wee Thanks for the suggestion! Alas, this doesn’t seem to fix the issue 😕

I believe it should be the same. We could be looking at a bug.

Oh interesting. Well, let me know if you have any other thoughts. I’ll keep poking around just in case (there are still some things I’m trying to explore through logging and spelunking the source code). Thanks!

Well, I turned your flow state handler into a task state handler and it worked in prefect server, although I expect that that is rather unwieldy since you have to specify it for each task. On the other hand, you're explicitly raising the signal for each task too, so it isn't that much more onerous.

@Amanda Wee You rock! The singleton approach does the trick! I should admit that it’s also quite possible I’m over-complicating things here. My end goal is to compile a slack alert that lists basic metadata about any mapped tasks that had a failure. My current real-world use case is very similar to the contrived division test we’re discussing here. I only need to log one minor bit of metadata that is, in fact, passed into the mapped task (it’s a U.S. state postal code). My goal is to not send notifications when a single task fails, but to send a single alert at the end of the flow run that notifies users about which U.S. states (not Prefect task/flow states) failed. It sounds like you’re suggesting I could simply inspect the inputs for all tasks to get this bit of metadata. That’d work fine, although I haven’t been able to figure out quite how to extract that info (I tried

flow.get_tasks

for example, but the

inputs

on the tasks appeared to be Prefect metadata as opposed to the actual values passed in as task arguments. Perhaps I’m missing something there? I should also mention that I’ll eventually be running this in parallel with the DaskExecutor, so not sure if that poses any potential landmines for the singleton approach. I’ll test that out with the LocalDaskExecutor to verify. Meantime, let me know if you think I can use a simpler

inputs

approach. And many many thanks for your help so far!

I didn’t read all of this yet but yes it is right that the result is different for local runs and Cloud/Server runs. This message contains the explanation

@Kevin Kho That’s super helpful! Thanks for the speedy reply! Is it still the recommended approach to query the GraphQL API for task state info?

Yes at the moment. Next release will have a simpler interface for it. Check this out for the query. Note that he had to include a 5 second wait because there is a delay of when the tasks get updated in the database. Might be more instant for server. That whole thread talks about sending a Slack message that contains whcih tasks fialed

@Kevin Kho Excellent! Thanks so much for all the helpful details. I’ll give the GraphQL approach a try. And thanks again to @Amanda Wee for all your help! 🙂

I’ll read this whole thread

@Kevin Kho OK. Greatly appreciated. Shout back if the flow code would be helpful and I can send that as well

Ok I’ll check with the team is state_result is expected to be NULL. Seems like the other fields updated already so it’s not that delay in writing to the database.

Ok. Thanks to you and the team for investigating. Really appreciate it!

Of course!

Wow, I guess I pinged at the right time! Maybe i’ll hold off on this until after I upgrade. But fwiw, this is also slightly an edge-casey nice-to-have on my part. Ultimately, i think it’d be fine for me to POST a slack alert saying a flow run failed (which I already have working), and then direct the user to Cloud UI to explore the details via logging panel. I was just trying to get a bit fancier with my Slack alert, which led me down this rabbit hole! 🙂 Anyhow, I’ll checkout 0.15.0 when it drops. Meantime, really appreciate you all being so responsive and please do pass along that so far I’m loving Prefect!

Of course! Thanks for your patience also!

@Kevin Kho Awesome. Thanks for the heads up!

I wouldn’t expect the output of the FlowRunView to be finished while the flow is still running (even at the state handler). Are you using this in the same flow run while grabbing the id from the context? The behavior for TaskRunView would be better. How are your TaskRunViews and can you show me your code?

Aha, ok, it sounds like i’m definitely using the flow run view incorrectly (or have the wrong expectations about it’s state). 1 sec, I’ll clean up the file and show the latest approach

Gotcha, I think this final notification should not be a task. It should be a

state handler

, and then the state handler will be attached to the flow. The

FlowRunView

will always be in a running state as the flow is running. The state handler will be triggered right before the flow is finishing, and the

state handler

will be responsible for the

TaskRunView

to get that info.

@Kevin Kho Aha. Ok, that makes sense. I tried the state handler approach with FlowRunView but it sounds like TaskRunView is how I should do approach this. I’ll check out the example you sent and really appreciate any more minimal example you can pass along. Thanks!

Is the purpose of this just when something fails, you want to figure out which one it was with a Slack message?

End goal is to be able to send a Slack alert when the run is completed. This is for a project that scrapes data from 50 US States. I’d like my alert message to contain a list of postal codes for the state scrapers that succeeded and failed. I’m trying to avoid slack alerts for each state scraper b/c the logging in Slack is too verbose. My goal is to somehow examine the state of all mapped tasks at end of the run. Then use the metadata from the succeeded results and metadata attached to failed signals to compile a single slack alert message.

Gotcha ok i’ll try to make something

You rock. I’ll keep tinkering as well. there are still a few things I haven’t tried, but really appreciate the sanity check and guidance!

I have been running into issues with TaskRunView I’ll need to ask the team about but I have a naive solution for you for now:

# Import Library
import prefect
from prefect import task, Flow
from datetime import timedelta
from prefect.backend.flow_run import FlowRunView, TaskRunView
from typing import Dict
import time

# I need this to hold the task inputs
class AnError(Exception):
    def __init__(self, message): 
        self.message = message
        super().__init__(self.message)
    
# Error Task
@task()
def scrape_state(state):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(str(state))
    if state == "C":
        raise AnError(f"Error Scraping {state}")
    else:
        return state

from prefect.triggers import always_run
@task(trigger=always_run)
def something_else(x):
    logger = prefect.context.get("logger")

    # x here will be a list of successes and errors
    successes = [_ for _ in x if not isinstance(_, Exception)]
    errors = [_.message for _ in x if isinstance(_, Exception)]
    errors = [_.split("Error Scraping ")[1] for _ in errors]

    <http://logger.info|logger.info>("successes")
    <http://logger.info|logger.info>(successes)
    <http://logger.info|logger.info>("errors")
    <http://logger.info|logger.info>(errors)
    return x

with Flow("flow_must_fail") as flow:
    states = ["A", "B", "C"]
    x = scrape_state.map(states)
    y = something_else(x, upstream_tasks = [x])

flow.run()

@Kevin Kho Aha, ok, that’s very clever! I was about to give up on this b/c it seemed that with my most recent script (that I just posted above), I would have to give up the benefits of full stack trace logging by raising signals in the task that (potentially) produces errors. But your approach seems to give me the best of all worlds. I’ll take it for a test spin this weekend. many thanks!!

Better version:

# Import Library
import prefect
from prefect import task, Flow
from datetime import timedelta
from prefect.backend.flow_run import FlowRunView, TaskRunView
from typing import Dict
import time

# I need this to hold the task inputs
class AnError(Exception):
    def __init__(self, message, input): 
        self.message = message
        self.input = input
        super().__init__(self.message)
    
# Error Task
@task()
def scrape_state(state):
    logger = prefect.context.get("logger")
    <http://logger.info|logger.info>(str(state))
    if state == "C":
        raise AnError(f"Error Scraping", input=state)
    else:
        return state

from prefect.triggers import always_run
@task(trigger=always_run)
def something_else(x):
    logger = prefect.context.get("logger")

    # x here will be a list of successes and errors
    successes = [_ for _ in x if not isinstance(_, Exception)]
    errors = [_.input for _ in x if isinstance(_, Exception)]

    <http://logger.info|logger.info>("successes")
    <http://logger.info|logger.info>(successes)
    <http://logger.info|logger.info>("errors")
    <http://logger.info|logger.info>(errors)
    return x

with Flow("flow_must_fail") as flow:
    states = ["A", "B", "C"]
    x = scrape_state.map(states)
    y = something_else(x, upstream_tasks = [x])

flow.run()

interesting. ok, i’ll try porting this approach to me real use case. Thanks again for all the help and let me know if you hear back any relevant details from the rest of the team about TaskRunViews. Have a great weekend!