Final Flow/Task State Warning

[toro-berlin]

First check

• I added a descriptive title to this issue.
• I used the GitHub search to find a similar request and didn't find it.
• I searched the Prefect documentation for this feature.

Prefect Version

2.x

Describe the current behavior

Situation: During a flow/task run, something suspicious happens that is no failure (code exception) or crash (infrastructure problem) but has to be reported and recognized by the operator.

Currently, as a flow developer, I cannot visually get the attention of the flow operator for a flow/task after execution and its warning message.

Example:

• Flow expects files in a given directory. But one day, there were no files there.
• We do not want to fail the task. Nothing failed in the flow, e.g., a retry will not fix the problem.
• We do not want to complete the task because something happened that needs the operator's attention.

Describe the proposed behavior

We want to return a warning state that can be caught and checked on.

Behavior:

• Color: Orange?
• No retries on flows/tasks with a warning.
• No Caching is happening for tasks with a warning.
• No further processing of downstream tasks that depend on the task in a warning state.

Example Use

from prefect.states import Warning
from prefect.client.schemas import State

@task
def return_warning() -> State:
    return Warning(message="Check the file input directory!")

Additional context

A different approach to get nearer to what I want is: Make retries applicable only to a subset of exceptions (#7576) to use an error state with a custom name.
But that will not expose the Warning State to the Cloud UI as a dedicated state, so it is not filterable etc...

[toro-berlin]

One thought that perhaps helps to get my understanding of the warning state better:

• Warning: Flow/task failed for a good reason that should be taken care of by the operator
• Failure: Flow/task failed for something else

[Ishankoradia]

@toro-berlin and @jakekaplan I am interested in contributing to this. Let me know if I can start on this or not.

[fatchat]

I'd like this feature too, our use-case is for dbt test failures - when a dbt test fails it's not a failure of the pipeline but the analysts who are responsible need to be aware that something is wrong with their data

[toro-berlin]

Hey @Ishankoradia , I am not working on that. But I suggest waiting for a go from the Prefect side.

[Ishankoradia]

Yup, will do that. Thanks @toro-berlin

[Ishankoradia]

@jakekaplan any word on this from your team ?

[jakekaplan]

Apologies for the delayed response! At this time I don't think we would accept a contribution for a brand new state.

However, I think you can accomplish what you want with a custom state name:

from prefect import flow
from prefect.states import State, StateType

@flow
def my_flow():
    Warning = State(type=StateType.CRASHED, name="WARNING")
    return Warning


if __name__ == '__main__':
    my_flow()

I'm using CRASHED as the state type here, because it fulfills a lot of the things you mentioned (no retries, no further processing, it's orange 😄)

[fatchat]

Thank you I'll try this

[toro-berlin]

Thanks for the idea, sounds like a great way to handle for the moment.
We will discuss it internally.

[Ishankoradia]

Thanks @jakekaplan , this definitely helps. So much less work now to handle the warning state.

[kyletl]

Apologies for the delayed response! At this time I don't think we would accept a contribution for a brand new state.

However, I think you can accomplish what you want with a custom state name:

from prefect import flow
from prefect.states import State, StateType

@flow
def my_flow():
    Warning = State(type=StateType.CRASHED, name="WARNING")
    return Warning


if __name__ == '__main__':
    my_flow()

I'm using CRASHED as the state type here, because it fulfills a lot of the things you mentioned (no retries, no further processing, it's orange 😄)

Hey @jakekaplan - one issue I'm running into with this solution is that crashed tasks don't seem to trigger downstream dependencies, so I can't have something finish in a "warn" state when I'd want it to trigger the next task. Any ideas on how to approach that?

My flow looks something like this, where I'd want the tasks to have a special case warning state if some of the upserts failed but there wasn't an error with the code execution or something like that

    t3 = upsert_objects.with_options(name="upsert_contacts").submit(
        object_name="contact", wait_for=[allow_failure(t2)]
    )

    t4 = upsert_objects.with_options(name="upsert_products").submit(
        object_name="product", wait_for=[allow_failure(t3)]
    )

[jakekaplan]

@kyletl it depends on what behavior you're looking for. It sounds like you may want a named COMPLETED state instead if you still want downstream dependencies to trigger. The requirements for the use case above we to explicitly not trigger downstream dependencies:

    Warning = State(type=StateType.COMPLETED, name="WARNING")

[kyletl]

@jakekaplan I ended up just using failed. What I really want is the orange color in the execution graph so that it's obvious from a glance that a special case was triggered but the task itself didn't fail.

[azjps]

Hi, this is also a major issue for us; it would be really helpful if we could encode and display completed states that are some spectrum between successful and failed (for example, partially failed, no-op, etc). Otherwise, a lot of key information is lost to viewers of the DAG.

As a (quite hacky) workaround, I noticed that in task_engine.py

prefect/src/prefect/utilities/engine.py

Line 272 in 42dcb00

if not state.is_completed() and not (

the task graph is not evaluated when upstream nodes are not is_completed() or is_failed(), but if one overrides is_completed() to True then the graph continues to evaluate, and then intermediate nodes can be displayed with the Cancelled & Crashed (and probably others) state colors on the UI:

from prefect.states import State, StateType

class CompletedState(State):
    def is_completed(self) -> bool:
        return True

    def is_crashed(self) -> bool:
        return False

return CompletedState(
    type=getattr(StateType, metadata.get("prefect_state_type")),  # e.g. CRASHED
    name=metadata.get("prefect_state_name"),  # e.g. SomeWarnings
    message=metadata.get("prefect_state_message", "Custom state"),
)

Obviously we would prefer an official solution, but it seems like this workaround would be enough for us (and hopefully others on this thread) to get by for the time being.